tests/test_rtsp.py

import pytest
from app import capture_rtsp_frames
from unittest.mock import patch, MagicMock
import cv2
import numpy as np

@patch('cv2.VideoCapture')
def test_rtsp_connection(mock_video_capture):
    """Test RTSP connection with mocked VideoCapture"""
    # Mock VideoCapture to simulate a successful connection
    mock_instance = mock_video_capture.return_value
    mock_instance.isOpened.return_value = True
    
    # Create a dummy frame
    dummy_frame = np.zeros((480, 640, 3), dtype=np.uint8)
    mock_instance.read.side_effect = [(True, dummy_frame), (False, None)]  # Simulate one frame then stop
    
    rtsp_url = "rtsp://test:8554/stream"
    frames = list(capture_rtsp_frames(rtsp_url, max_frames=1))  # Consume generator
    
    assert len(frames) == 1, "Expected one frame from mocked RTSP stream"
    assert frames[0][0] is not None, "Frame should not be None"
    assert frames[0][1] is not None, "Timestamp should not be None"
    
    mock_video_capture.assert_called_with(rtsp_url)
    mock_instance.release.assert_called_once()

def test_rtsp_connection_failure():
    """Test RTSP connection failure handling"""
    with patch('cv2.VideoCapture') as mock_video_capture:
        mock_instance = mock_video_capture.return_value
        mock_instance.isOpened.return_value = False
        
        rtsp_url = "rtsp://invalid:8554/stream"
        
        with pytest.raises(ValueError, match="RTSP stream not accessible"):
            list(capture_rtsp_frames(rtsp_url, max_frames=1))