4eeddc4b1f
Modular async pipeline: FrameSource->Vision->context and STT/text->Brain->TTS. All stages are Protocols; mock backends run end-to-end with no deps/keys. Real backends included: mss screen capture, Claude vision+brain (guarded imports).
49 lines
1.2 KiB
Python
49 lines
1.2 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_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")
|