75 lines
2.3 KiB
Python
75 lines
2.3 KiB
Python
import json
|
|
import time
|
|
|
|
import engine_llm
|
|
from engine_llm import (
|
|
filler_due,
|
|
on_playback_interrupted,
|
|
parse_metadata,
|
|
parse_sse_line,
|
|
sanitize_for_tts,
|
|
truncate_spoken,
|
|
validate_metadata,
|
|
)
|
|
|
|
|
|
def test_parse_sse_line():
|
|
event = parse_sse_line('data: {"type":"text_delta","text":"Hi"}')
|
|
assert event == {"type": "text_delta", "text": "Hi"}
|
|
assert parse_sse_line(": heartbeat") is None
|
|
|
|
|
|
def test_filler_waits_until_the_delay():
|
|
started = time.time()
|
|
assert filler_due(started, started + 0.2, False, False, True, 800) is False
|
|
assert filler_due(started, started + 0.9, False, False, True, 800) is True
|
|
assert filler_due(started, started + 0.9, True, False, True, 800) is False
|
|
|
|
|
|
def test_cancel_reports_only_the_spoken_prefix():
|
|
class Client:
|
|
def __init__(self):
|
|
self.closed = False
|
|
self.spoken = None
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def mark_interrupted(self, session_id, spoken):
|
|
self.spoken = (session_id, spoken)
|
|
|
|
client = Client()
|
|
spoken = on_playback_interrupted(client, "s1", "Hello there friend", 0.5)
|
|
assert client.closed is True
|
|
assert spoken == truncate_spoken("Hello there friend", 0.5)
|
|
assert client.spoken[0] == "s1"
|
|
assert len(client.spoken[1]) < len("Hello there friend")
|
|
|
|
|
|
def test_metadata_requires_a_session():
|
|
assert parse_metadata("") is None
|
|
assert parse_metadata(json.dumps({"bookId": "3"})) is None
|
|
data = validate_metadata(parse_metadata(json.dumps({"sessionId": "s1", "voiceId": "Joanna"})))
|
|
assert data["sessionId"] == "s1"
|
|
try:
|
|
validate_metadata({})
|
|
except ValueError as error:
|
|
assert "sessionId" in str(error)
|
|
else:
|
|
raise AssertionError("expected ValueError")
|
|
|
|
|
|
def test_sanitizer_strips_markdown():
|
|
assert sanitize_for_tts("**Hello** `code`") == "Hello code"
|
|
assert sanitize_for_tts("- Keep going") == "Keep going"
|
|
|
|
|
|
def test_sanitizer_strips_a_marker_split_across_deltas():
|
|
before = engine_llm.tts_sanitizer_hits
|
|
left = sanitize_for_tts("The dose is [[p:")
|
|
right = sanitize_for_tts("47]] fifty milligrams.")
|
|
spoken = f"{left} {right}"
|
|
assert "[[" not in spoken
|
|
assert "]]" not in spoken
|
|
assert engine_llm.tts_sanitizer_hits > before
|