"""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, }