30 lines
808 B (Stored with Git LFS)
Python
30 lines
808 B (Stored with Git LFS)
Python
import pytest
|
|
import cv2
|
|
|
|
import numpy as np
|
|
from av import VideoFrame
|
|
from src.api.streaming.video_source import OpenCV_Video_Source_Async, StreamTrackOpencv
|
|
|
|
pytest_plugins = ('pytest_asyncio',)
|
|
|
|
@pytest.mark.timeout(500)
|
|
@pytest.mark.asyncio
|
|
async def test_async_video_source():
|
|
v_src = cv2.VideoCapture("test/video/Big_Buck_Bunny_1080_10s_10MB.mp4")
|
|
|
|
def opencv_frame_func():
|
|
return cv2.VideoCapture.read(v_src)[1]
|
|
|
|
video_source = OpenCV_Video_Source_Async(opencv_frame_func)
|
|
stream_track = StreamTrackOpencv(video_source)
|
|
video_source.start()
|
|
|
|
for i in range(120):
|
|
#frame = await video_source.get_frame()
|
|
frame = await stream_track.recv()
|
|
assert frame is not None
|
|
assert type(frame) is VideoFrame
|
|
|
|
video_source.stop()
|
|
|