STT-Assistant/tests/test_events.py

"""Tests for the NDJSON event envelope and codec."""

import json

import pytest

from stt_daemon import events


def test_make_event_envelope():
    event = events.make_event(events.HEARTBEAT, 7, uptime_s=12.5)
    assert event["v"] == events.PROTOCOL_VERSION
    assert event["type"] == events.HEARTBEAT
    assert event["seq"] == 7
    assert isinstance(event["ts"], float)
    assert event["uptime_s"] == 12.5


def test_make_event_explicit_timestamp():
    event = events.make_event(events.STATUS, 1, ts=1000.5)
    assert event["ts"] == 1000.5


@pytest.mark.parametrize("type_", sorted(events.ALL_TYPES))
def test_round_trip_every_type(type_):
    event = events.make_event(type_, 42, payload={"á": "ñ"}, n=1)
    line = events.encode(event)
    assert line.endswith(b"\n")
    assert events.decode(line) == event


def test_encode_is_utf8_not_escaped():
    line = events.encode(events.make_event(events.TRANSCRIPT, 1, text="café"))
    assert "café".encode("utf-8") in line


def test_decode_accepts_str_and_bytes():
    event = events.make_event(events.SPEECH_START, 3)
    assert events.decode(events.encode(event)) == event
    assert events.decode(events.encode(event).decode("utf-8")) == event


@pytest.mark.parametrize(
    "bad",
    [
        b"not json\n",
        b"[1,2,3]\n",
        b'"a string"\n',
        b"42\n",
        b'{"type":"heartbeat","seq":1,"ts":1.0}\n',  # missing v
        b'{"v":1,"seq":1,"ts":1.0}\n',  # missing type
        b'{"v":1,"type":"heartbeat","ts":1.0}\n',  # missing seq
        b'{"v":1,"type":"heartbeat","seq":1}\n',  # missing ts
        b"\xff\xfe\n",  # invalid utf-8
    ],
)
def test_decode_rejects_malformed(bad):
    with pytest.raises(events.DecodeError):
        events.decode(bad)


def test_decode_rejects_wrong_argument_type():
    with pytest.raises(events.DecodeError):
        events.decode(json.dumps([]).encode() + b"\n")  # object required


def test_all_types_is_complete():
    assert events.ALL_TYPES == {
        events.TRANSCRIPT,
        events.SEGMENT_DONE,
        events.SPEECH_START,
        events.SPEECH_STOP,
        events.HEARTBEAT,
        events.STATUS,
        events.RING_END,
        events.SPEAKERS,
    }


def test_speakers_event_constant():
    assert events.SPEAKERS == "speakers"
    assert events.SPEAKERS in events.ALL_TYPES