Files
watch_sceen_ai/tests/test_pipeline.py
EJClaw 2b6059141e feat(voice): run pipeline eyes-free (STT -> Brain -> TTS), defer screen share
Make source/vision optional so the conversation loop runs with no screen
capture. Add Settings.voice() preset and `python -m wsai --voice` demo.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-08-11 00:05:02 +09:00

70 lines
1.9 KiB
Python

"""Smoke test: the mock pipeline must run end-to-end and route screen context
into the brain's replies."""
import asyncio
from wsai.backends.mock import (
MockBrain,
MockFrameSource,
MockSTT,
MockTTS,
MockVision,
)
from wsai.pipeline import Pipeline
def test_mock_pipeline_runs_and_replies(capsys):
replies: list[str] = []
class CapturingTTS(MockTTS):
async def speak(self, reply):
replies.append(reply.text)
pipe = Pipeline(
source=MockFrameSource(interval=0.05, limit=3),
vision=MockVision(),
brain=MockBrain(),
stt=MockSTT(script=["화면에 뭐 보여?"], interval=0.1),
tts=CapturingTTS(),
)
asyncio.run(asyncio.wait_for(pipe.run(), timeout=5))
assert replies, "brain produced no reply"
# The reply must embed the screen observation → context reached the brain.
assert "화면:" in replies[0]
def test_voice_only_pipeline_runs_without_eyes():
"""Eyes-free config (no source/vision) still runs STT -> Brain -> TTS."""
replies: list[str] = []
class CapturingTTS(MockTTS):
async def speak(self, reply):
replies.append(reply.text)
pipe = Pipeline(
brain=MockBrain(),
stt=MockSTT(script=["안녕", "잘 있어"], interval=0.05),
tts=CapturingTTS(),
)
asyncio.run(asyncio.wait_for(pipe.run(), timeout=5))
assert len(replies) == 2, "voice loop did not reply to every utterance"
# No eyes → the brain must report it has not seen a screen.
assert "아직 화면을 못 읽었어요" in replies[0]
def test_history_is_bounded():
pipe = Pipeline(
source=MockFrameSource(limit=0),
vision=MockVision(),
brain=MockBrain(),
history_turns=3,
)
for i in range(10):
pipe._remember(f"u{i}", f"a{i}")
assert len(pipe._history) == 3
assert pipe._history[-1] == ("u9", "a9")