mirror of
https://github.com/ApfelTeeSaft/restream_playout.git
synced 2026-08-26 19:33:32 +00:00
28 lines
699 B
Python
28 lines
699 B
Python
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
from restream_playout.queue import VideoQueue
|
|
|
|
|
|
class QueueTests(unittest.TestCase):
|
|
def test_fifo_and_wait(self) -> None:
|
|
queue = VideoQueue()
|
|
|
|
def producer() -> None:
|
|
time.sleep(0.02)
|
|
queue.enqueue_many(["a.mp4", "b.mp4"])
|
|
|
|
thread = threading.Thread(target=producer)
|
|
thread.start()
|
|
self.assertTrue(queue.wait(1))
|
|
self.assertEqual(str(queue.peek()), "a.mp4")
|
|
self.assertEqual(str(queue.pop()), "a.mp4")
|
|
self.assertEqual(queue.size(), 1)
|
|
queue.clear()
|
|
self.assertEqual(queue.size(), 0)
|
|
thread.join()
|
|
|