206 lines
6.6 KiB
Python
206 lines
6.6 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Tests for the voice-regeneration queue.
|
|
|
|
The queue exists because regenerating a voice used to be one-at-a-time: the
|
|
route backgrounded the work but refused a second request with a 409, and the UI
|
|
had no per-speaker state to show. These cover the queue semantics the reviewer
|
|
depends on - enqueue-many, run-serially, report per item - without invoking TTS.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from umbraaudio.voices import voice_queue as vq
|
|
|
|
|
|
def _make(tmp_path: Path) -> vq.VoiceQueue:
|
|
return vq.VoiceQueue(tmp_path, SimpleNamespace(tts=SimpleNamespace()))
|
|
|
|
|
|
@pytest.fixture
|
|
def queue(tmp_path: Path) -> vq.VoiceQueue:
|
|
"""A queue whose worker is disabled, so items stay ``queued``.
|
|
|
|
Starting a real worker would drain the queue on a background thread and make
|
|
these assertions a race. The enqueue/status/persist logic under test is
|
|
untouched; the drain itself is covered separately by
|
|
``test_worker_drains_items_in_order``.
|
|
"""
|
|
q = _make(tmp_path)
|
|
q._ensure_worker = lambda: None # type: ignore[method-assign]
|
|
return q
|
|
|
|
|
|
@pytest.fixture
|
|
def draining(tmp_path: Path) -> vq.VoiceQueue:
|
|
"""A queue with a stubbed TTS call but a real worker, to exercise the drain."""
|
|
q = _make(tmp_path)
|
|
|
|
def _no_generation(self: vq.VoiceQueue, item: vq.QueueItem) -> None:
|
|
with self._lock:
|
|
item.status = vq.STATUS_DONE
|
|
item.finished_at = 0.0
|
|
self._save()
|
|
|
|
q._run_item = _no_generation.__get__(q, vq.VoiceQueue) # type: ignore[method-assign]
|
|
return q
|
|
|
|
|
|
def test_worker_drains_items_in_order(draining: vq.VoiceQueue) -> None:
|
|
"""The worker must consume queued items one at a time, in enqueue order."""
|
|
import time
|
|
|
|
for name in ("one", "two", "three"):
|
|
draining.enqueue(name)
|
|
|
|
deadline = time.time() + 10
|
|
while draining.is_busy() and time.time() < deadline:
|
|
time.sleep(0.05)
|
|
|
|
statuses = [i["status"] for i in draining.snapshot()["items"]]
|
|
assert statuses == [vq.STATUS_DONE] * 3
|
|
assert draining.is_busy() is False
|
|
|
|
|
|
# ------------------------------------------------------------------ enqueue
|
|
|
|
|
|
def test_enqueue_returns_queued(queue: vq.VoiceQueue) -> None:
|
|
item = queue.enqueue("Manfred")
|
|
assert item["status"] == vq.STATUS_QUEUED
|
|
assert item["speaker_name"] == "Manfred"
|
|
|
|
|
|
def test_enqueue_is_idempotent_for_a_pending_speaker(queue: vq.VoiceQueue) -> None:
|
|
"""A double-click must not queue the same regeneration twice."""
|
|
queue.enqueue("Manfred")
|
|
before = queue.snapshot()["total"]
|
|
queue.enqueue("Manfred")
|
|
assert queue.snapshot()["total"] == before
|
|
|
|
|
|
def test_enqueue_can_queue_more_than_one_at_a_time(queue: vq.VoiceQueue) -> None:
|
|
"""The whole point: a second speaker is accepted while the first runs."""
|
|
for name in ("A", "B", "C"):
|
|
assert queue.enqueue(name)["status"] == vq.STATUS_QUEUED
|
|
assert queue.snapshot()["total"] == 3
|
|
|
|
|
|
def test_enqueue_preserves_order(queue: vq.VoiceQueue) -> None:
|
|
for name in ("first", "second", "third"):
|
|
queue.enqueue(name)
|
|
order = [i["speaker_name"] for i in queue.snapshot()["items"]]
|
|
assert order == ["first", "second", "third"]
|
|
|
|
|
|
# ------------------------------------------------------------------- status
|
|
|
|
|
|
def test_status_for_unknown_speaker_is_none(queue: vq.VoiceQueue) -> None:
|
|
assert queue.status_for("nobody") is None
|
|
|
|
|
|
def test_snapshot_counts_only_pending_as_queued(queue: vq.VoiceQueue) -> None:
|
|
queue.enqueue("A")
|
|
queue.enqueue("B")
|
|
snap = queue.snapshot()
|
|
assert snap["queued"] == 2
|
|
assert snap["running"] is None
|
|
assert snap["total"] == 2
|
|
|
|
|
|
def test_is_busy_reflects_pending_work(queue: vq.VoiceQueue) -> None:
|
|
assert queue.is_busy() is False
|
|
queue.enqueue("A")
|
|
assert queue.is_busy() is True
|
|
|
|
|
|
# ------------------------------------------------------------------- cancel
|
|
|
|
|
|
def test_cancel_removes_a_queued_item(queue: vq.VoiceQueue) -> None:
|
|
queue.enqueue("A")
|
|
assert queue.cancel("A") is True
|
|
assert queue.status_for("A") is None
|
|
|
|
|
|
def test_cancel_refuses_a_running_item(queue: vq.VoiceQueue) -> None:
|
|
queue.enqueue("A")
|
|
with queue._lock:
|
|
queue._items[0].status = vq.STATUS_RUNNING
|
|
assert queue.cancel("A") is False
|
|
|
|
|
|
# -------------------------------------------------------------- persistence
|
|
|
|
|
|
def test_queue_state_is_persisted(tmp_path: Path) -> None:
|
|
cfg = SimpleNamespace(tts=SimpleNamespace())
|
|
q = vq.VoiceQueue(tmp_path, cfg)
|
|
q.enqueue("Persisted")
|
|
reloaded = vq.VoiceQueue(tmp_path, cfg)
|
|
assert reloaded.status_for("Persisted") is not None
|
|
|
|
|
|
def test_running_item_becomes_error_after_restart(tmp_path: Path) -> None:
|
|
"""A process that dies mid-generation must not leave a stuck 'running' tag."""
|
|
cfg = SimpleNamespace(tts=SimpleNamespace())
|
|
q = vq.VoiceQueue(tmp_path, cfg)
|
|
q._items.append(vq.QueueItem(speaker="Ghost", status=vq.STATUS_RUNNING))
|
|
q._save()
|
|
|
|
reloaded = vq.VoiceQueue(tmp_path, cfg)
|
|
ghost = reloaded.status_for("Ghost")
|
|
assert ghost["status"] == vq.STATUS_ERROR
|
|
assert "Interrupt" in ghost["error"]
|
|
|
|
|
|
def test_corrupt_status_file_does_not_crash(tmp_path: Path) -> None:
|
|
cfg = SimpleNamespace(tts=SimpleNamespace())
|
|
vq.VoiceQueue(tmp_path, cfg)
|
|
(tmp_path / "voice_queue.json").write_text("{not json")
|
|
assert vq.VoiceQueue(tmp_path, cfg).snapshot()["total"] == 0
|
|
|
|
|
|
def test_finished_history_is_trimmed(tmp_path: Path) -> None:
|
|
"""Terminal items must not accumulate without bound."""
|
|
cfg = SimpleNamespace(tts=SimpleNamespace())
|
|
q = vq.VoiceQueue(tmp_path, cfg)
|
|
for i in range(80):
|
|
q._items.append(
|
|
vq.QueueItem(speaker=f"old{i}", status=vq.STATUS_DONE)
|
|
)
|
|
q._trim_history()
|
|
assert len([i for i in q._items if i.status in vq.TERMINAL_STATES]) <= 50
|
|
|
|
|
|
# --------------------------------------------------------------- vocabulary
|
|
|
|
|
|
def test_status_vocabulary_matches_the_controller_mapping() -> None:
|
|
"""These names are what the controller will map onto node task states."""
|
|
assert vq.STATUS_QUEUED == "queued"
|
|
assert vq.STATUS_RUNNING == "running"
|
|
assert vq.STATUS_DONE == "done"
|
|
assert vq.STATUS_ERROR == "error"
|
|
assert {vq.STATUS_DONE, vq.STATUS_ERROR} == vq.TERMINAL_STATES
|
|
|
|
|
|
def test_public_state_excludes_internal_timings() -> None:
|
|
item = vq.QueueItem(speaker="X", row_index=4)
|
|
assert set(item.public()) == {
|
|
"speaker",
|
|
"speaker_name",
|
|
"status",
|
|
"error",
|
|
"row_index",
|
|
}
|