"""Tests for OutputPipeline filter chaining."""
import pytest
from typing import List
class MockFilter:
"""Mock filter for testing pipeline behavior."""
def __init__(
self, transform: str = "", should_process_result: bool = True, name: str = ""
):
"""Initialize mock filter.
Args:
transform: String to append to input text
should_process_result: What should_process() returns
name: Name for debugging
"""
self.transform = transform
self.should_process_result = should_process_result
self.name = name
self.filter_called = False
self.should_process_called = False
def filter(self, text: str) -> str:
"""Apply transformation."""
self.filter_called = True
if self.transform:
return f"{text}{self.transform}"
return text
def should_process(self, text: str) -> bool:
"""Return configured should_process result."""
self.should_process_called = True
return self.should_process_result
class TestOutputPipeline:
"""Tests for OutputPipeline class."""
def test_empty_pipeline_returns_input(self):
"""Empty pipeline returns input unchanged."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
pipeline = OutputPipeline([])
result = pipeline.process("hello world")
assert result == "hello world"
def test_single_filter_applied(self):
"""Single filter is applied to input."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock = MockFilter(transform="!")
pipeline = OutputPipeline([mock])
result = pipeline.process("hello")
assert result == "hello!"
assert mock.filter_called
def test_filters_chained_in_order(self):
"""Filters are applied in order."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
mock3 = MockFilter(transform="-C")
pipeline = OutputPipeline([mock1, mock2, mock3])
result = pipeline.process("test")
assert result == "test-A-B-C"
def test_should_process_respected(self):
"""Filters with should_process=False are skipped."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B", should_process_result=False)
mock3 = MockFilter(transform="-C")
pipeline = OutputPipeline([mock1, mock2, mock3], respect_should_process=True)
result = pipeline.process("test")
assert result == "test-A-C"
assert mock1.filter_called
assert not mock2.filter_called # Skipped
assert mock3.filter_called
def test_respect_should_process_disabled(self):
"""All filters applied when respect_should_process is False."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B", should_process_result=False)
pipeline = OutputPipeline([mock1, mock2], respect_should_process=False)
result = pipeline.process("test")
assert result == "test-A-B"
assert mock1.filter_called
assert mock2.filter_called
def test_empty_text_returns_empty(self):
"""Empty text input returns empty string."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock = MockFilter(transform="!")
pipeline = OutputPipeline([mock])
result = pipeline.process("")
assert result == ""
assert not mock.filter_called
def test_none_filters_uses_empty_list(self):
"""Pipeline with None filters defaults to empty list."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
pipeline = OutputPipeline(None)
result = pipeline.process("hello")
assert result == "hello"
assert len(pipeline) == 0
def test_add_filter(self):
"""add_filter appends to pipeline."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
pipeline = OutputPipeline([mock1])
pipeline.add_filter(mock2)
result = pipeline.process("test")
assert result == "test-A-B"
assert len(pipeline) == 2
def test_insert_filter(self):
"""insert_filter inserts at specified position."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
mock3 = MockFilter(transform="-C")
pipeline = OutputPipeline([mock1, mock3])
pipeline.insert_filter(1, mock2)
result = pipeline.process("test")
assert result == "test-A-B-C"
def test_remove_filter(self):
"""remove_filter removes specified filter."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
pipeline = OutputPipeline([mock1, mock2])
pipeline.remove_filter(mock1)
result = pipeline.process("test")
assert result == "test-B"
assert len(pipeline) == 1
def test_remove_nonexistent_filter(self):
"""remove_filter ignores filter not in pipeline."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
pipeline = OutputPipeline([mock1])
pipeline.remove_filter(mock2) # Not in pipeline
assert len(pipeline) == 1
def test_clear(self):
"""clear removes all filters."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
pipeline = OutputPipeline([mock1, mock2])
pipeline.clear()
result = pipeline.process("test")
assert result == "test"
assert len(pipeline) == 0
def test_len(self):
"""__len__ returns filter count."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
pipeline = OutputPipeline([MockFilter(), MockFilter(), MockFilter()])
assert len(pipeline) == 3
def test_method_chaining(self):
"""Pipeline modification methods support chaining."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
mock1 = MockFilter(transform="-A")
mock2 = MockFilter(transform="-B")
pipeline = (
OutputPipeline()
.add_filter(mock1)
.add_filter(mock2)
)
result = pipeline.process("test")
assert result == "test-A-B"
class TestOutputPipelineWithRealFilters:
"""Tests using actual RoleplayFilter and TTSFilter."""
def test_roleplay_filter_in_pipeline(self):
"""RoleplayFilter works in pipeline."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
pipeline = OutputPipeline([RoleplayFilter()])
result = pipeline.process('*sighs* "Hello there"')
assert result == "Hello there"
def test_roleplay_then_tts_chain(self):
"""RoleplayFilter chains with TTSFilter."""
from src.localvoicemode.speech.output_pipeline import OutputPipeline
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
from src.localvoicemode.speech.filter import TTSFilter
pipeline = OutputPipeline(
[RoleplayFilter(), TTSFilter()], respect_should_process=False
)
# Roleplay extracts dialogue, TTSFilter cleans it
result = pipeline.process('*action* "Hello world"')
assert "Hello world" in result
class TestCreatePipelineForSkill:
"""Tests for create_pipeline_for_skill factory function."""
def test_no_skill_returns_tts_only(self):
"""No skill argument returns TTSFilter only."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
pipeline = create_pipeline_for_skill(None)
assert len(pipeline) == 1
def test_skill_without_filters_returns_tts_only(self):
"""Skill with no output_filters returns TTSFilter only."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
skill = Skill(
id="test",
name="Test",
display_name="Test",
description="Test skill",
system_prompt="You are a test.",
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 1
def test_skill_with_roleplay_filter(self):
"""Skill with roleplay filter includes RoleplayFilter."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
skill = Skill(
id="rp",
name="Roleplay",
display_name="Roleplay",
description="Roleplay skill",
system_prompt="You are roleplaying.",
output_filters=["roleplay"],
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 1
assert isinstance(pipeline.filters[0], RoleplayFilter)
def test_skill_with_tts_filter(self):
"""Skill with tts filter includes TTSFilter."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
from src.localvoicemode.speech.filter import TTSFilter
skill = Skill(
id="assistant",
name="Assistant",
display_name="Assistant",
description="Assistant skill",
system_prompt="You are an assistant.",
output_filters=["tts"],
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 1
assert isinstance(pipeline.filters[0], TTSFilter)
def test_skill_with_roleplay_and_tts(self):
"""Skill with both filters chains them in order."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
from src.localvoicemode.speech.filter import TTSFilter
skill = Skill(
id="rp-char",
name="RP Character",
display_name="RP Character",
description="Roleplay character",
system_prompt="You are a character.",
output_filters=["roleplay", "tts"],
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 2
assert isinstance(pipeline.filters[0], RoleplayFilter)
assert isinstance(pipeline.filters[1], TTSFilter)
def test_skill_with_unknown_filter_ignored(self):
"""Unknown filter names are silently ignored."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
skill = Skill(
id="test",
name="Test",
display_name="Test",
description="Test skill",
system_prompt="Test.",
output_filters=["unknown", "also_unknown"],
)
pipeline = create_pipeline_for_skill(skill)
# Falls back to TTS only
assert len(pipeline) == 1
def test_skill_with_mixed_known_unknown(self):
"""Known filters used, unknown ignored."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
skill = Skill(
id="test",
name="Test",
display_name="Test",
description="Test skill",
system_prompt="Test.",
output_filters=["unknown", "roleplay", "also_unknown"],
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 1
assert isinstance(pipeline.filters[0], RoleplayFilter)
def test_filter_name_case_insensitive(self):
"""Filter names are case insensitive."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
from src.localvoicemode.speech.roleplay_filter import RoleplayFilter
from src.localvoicemode.speech.filter import TTSFilter
skill = Skill(
id="test",
name="Test",
display_name="Test",
description="Test skill",
system_prompt="Test.",
output_filters=["ROLEPLAY", "TTS"],
)
pipeline = create_pipeline_for_skill(skill)
assert len(pipeline) == 2
assert isinstance(pipeline.filters[0], RoleplayFilter)
assert isinstance(pipeline.filters[1], TTSFilter)
def test_pipeline_processes_text(self):
"""Created pipeline correctly processes text."""
from src.localvoicemode.speech.output_pipeline import create_pipeline_for_skill
from src.localvoicemode.skills.skill import Skill
skill = Skill(
id="rp",
name="Roleplay",
display_name="Roleplay",
description="Roleplay skill",
system_prompt="You are roleplaying.",
output_filters=["roleplay"],
)
pipeline = create_pipeline_for_skill(skill)
result = pipeline.process('*action* "Dialogue here"')
assert result == "Dialogue here"
