STT-Assistant/tests/test_assistant_integration.py

"""Integration: the assistant service consuming a real publish-only bus."""

import os
import shutil
import tempfile
import time

import pytest

from stt_daemon import events
from stt_daemon.assistant.service import AssistantService
from stt_daemon.bus import EventPublisher, EventSubscriber
from stt_daemon.config import Config


@pytest.fixture
def sock_path():
    directory = tempfile.mkdtemp(prefix="sttb-")
    try:
        yield os.path.join(directory, "s.sock")
    finally:
        shutil.rmtree(directory, ignore_errors=True)


def _wait(predicate, timeout=5.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        if predicate():
            return True
        time.sleep(0.01)
    return predicate()


def _heartbeat_fields():
    return {
        "in_flight": False,
        "segments_done": 0,
        "worker_stall_s": 0.0,
        "queue_depth": 0,
        "frames_since_last": 1,
        "rms": 0.0,
        "capture": "test",
        "device_ok": True,
        "model_loaded": True,
        "uptime_s": 1.0,
        "last_transcript_age_s": 0.0,
    }


def test_service_consumes_real_bus(sock_path):
    cfg = Config()
    publisher = EventPublisher(sock_path, ring_size=10)
    publisher.start()

    subscriber = EventSubscriber(sock_path, reconnect_delay=0.05)
    emitted: list[dict] = []
    service = AssistantService(
        cfg,
        subscriber,
        emit=lambda type_, **fields: emitted.append({"type": type_, **fields}),
    )
    service.start()

    try:
        publisher.emit(events.HEARTBEAT, **_heartbeat_fields())
        assert _wait(lambda: service.stt_ok() is True)

        assert service.arm() is True
        assert _wait(lambda: any(e.get("mode") == "command_armed" for e in emitted))

        publisher.emit(events.SPEECH_START)
        assert _wait(lambda: service.speech_open is True)

        publisher.emit(events.SPEECH_STOP, seg_id=5)
        assert _wait(lambda: service.awaited_seg_id == 5)
        assert service.ready() is False

        publisher.emit(events.SEGMENT_DONE, seg_id=5, emitted=True)
        assert _wait(lambda: service.ready() is True)

        publisher.emit(events.TRANSCRIPT, seg_id=5, text="hello", lang="en", lang_prob=0.9)
        assert _wait(lambda: service.command_text == "hello")

        assert service.execute() == "hello"
        assert service.finish() is True
    finally:
        service.stop()
        publisher.close()