38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
import queue, threading, importlib
|
|
import whisper_app.app as app_mod
|
|
|
|
def setup_function():
|
|
importlib.reload(app_mod)
|
|
|
|
def test_log_before_queue_buffers():
|
|
app_mod.log("hello")
|
|
app_mod.log("world")
|
|
assert app_mod._log_buffer == ["hello", "world"]
|
|
assert app_mod._log_queue is None
|
|
|
|
def test_log_buffer_capped_at_500():
|
|
for i in range(600):
|
|
app_mod.log(str(i))
|
|
assert len(app_mod._log_buffer) <= 500
|
|
|
|
def test_set_log_queue_flushes_buffer():
|
|
app_mod.log("buffered")
|
|
q = queue.Queue()
|
|
app_mod.set_log_queue(q)
|
|
assert q.get_nowait() == "buffered"
|
|
assert app_mod._log_buffer == []
|
|
|
|
def test_log_after_queue_goes_to_queue():
|
|
q = queue.Queue()
|
|
app_mod.set_log_queue(q)
|
|
app_mod.log("direct")
|
|
assert q.get_nowait() == "direct"
|
|
|
|
def test_log_thread_safe():
|
|
q = queue.Queue()
|
|
app_mod.set_log_queue(q)
|
|
threads = [threading.Thread(target=lambda: app_mod.log("x")) for _ in range(50)]
|
|
for t in threads: t.start()
|
|
for t in threads: t.join()
|
|
assert q.qsize() == 50
|