3cc262ed18
Pipeline.run() used asyncio.gather, so if one loop raised, the failing coroutine propagated while the sibling loops kept running detached; aclose() in the finally then closed a source/stt out from under a still-live loop. Switch to asyncio.TaskGroup so a failing loop cancels+awaits the siblings before teardown. Add a regression test asserting an error in the conversation loop cancels the perception loop and still closes every source.
111 lines
3.2 KiB
Python
111 lines
3.2 KiB
Python
"""Smoke test: the mock pipeline must run end-to-end and route screen context
|
|
into the brain's replies."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from wsai.backends.mock import (
|
|
MockBrain,
|
|
MockFrameSource,
|
|
MockSTT,
|
|
MockTTS,
|
|
MockVision,
|
|
)
|
|
from wsai.interfaces import Frame
|
|
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_error_in_one_loop_cancels_siblings_and_closes():
|
|
"""If the conversation loop raises, the perception loop must be cancelled
|
|
(not left running detached) and every source must still be closed — i.e. no
|
|
close-during-use and no orphaned task."""
|
|
closed = {"source": False, "stt": False}
|
|
|
|
class ForeverSource:
|
|
async def frames(self):
|
|
while True:
|
|
await asyncio.sleep(0.01)
|
|
yield Frame(data=b"", width=1, height=1, ts=0.0)
|
|
|
|
async def aclose(self):
|
|
closed["source"] = True
|
|
|
|
class BoomSTT(MockSTT):
|
|
async def aclose(self):
|
|
closed["stt"] = True
|
|
|
|
class BoomBrain(MockBrain):
|
|
async def respond(self, user_text, screen, history):
|
|
raise RuntimeError("boom")
|
|
|
|
pipe = Pipeline(
|
|
source=ForeverSource(),
|
|
vision=MockVision(),
|
|
brain=BoomBrain(),
|
|
stt=BoomSTT(script=["hi"], interval=0.01),
|
|
tts=MockTTS(),
|
|
)
|
|
|
|
with pytest.raises(BaseException): # TaskGroup raises an ExceptionGroup
|
|
asyncio.run(asyncio.wait_for(pipe.run(), timeout=5))
|
|
|
|
assert closed["source"] is True, "perception source was not closed (orphaned loop)"
|
|
assert closed["stt"] is True, "stt was not closed"
|
|
|
|
|
|
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")
|