watch_sceen_ai/tests/test_pipeline.py

"""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")


def test_prewarm_warms_backends_and_survives_failure():
    """Backends exposing warmup() are preloaded at startup; a warmup that
    raises is logged but must not abort the run (first-utterance latency is an
    optimization, not a hard requirement)."""
    warmed: list[str] = []

    class WarmTTS(MockTTS):
        async def warmup(self):
            warmed.append("tts")

    class BoomWarmSTT(MockSTT):
        async def warmup(self):
            warmed.append("stt")
            raise RuntimeError("model unavailable")

    pipe = Pipeline(
        brain=MockBrain(),
        stt=BoomWarmSTT(script=["안녕"], interval=0.01),
        tts=WarmTTS(),
    )

    asyncio.run(asyncio.wait_for(pipe.run(), timeout=5))

    assert "tts" in warmed and "stt" in warmed, "warmup() not called on backends"