STT-Assistant/tests/test_daemon.py

"""Tests for the daemon's transcription backlog warning."""

import logging

from stt_daemon.config import Config
from stt_daemon.daemon import Daemon


def _daemon(threshold=2, queue_size=10):
    cfg = Config()
    cfg.daemon.backlog_warn_at = threshold
    cfg.daemon.queue_size = queue_size
    return Daemon(cfg)


def test_queue_size_comes_from_config():
    daemon = _daemon(queue_size=7)
    assert daemon.segment_queue.maxsize == 7


def test_backlog_warning_is_edge_triggered(caplog):
    daemon = _daemon(threshold=2)

    with caplog.at_level(logging.WARNING, logger="stt_daemon"):
        daemon.segment_queue.put("a")
        daemon._check_backlog()
        assert "backlog" not in caplog.text.lower()

        daemon.segment_queue.put("b")
        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 1

        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 1

        daemon.segment_queue.get()
        daemon.segment_queue.get()
        daemon._check_backlog()
        assert daemon._backlog_warned is False

        daemon.segment_queue.put("c")
        daemon.segment_queue.put("d")
        daemon._check_backlog()
        assert caplog.text.lower().count("backlog") == 2