security-ai-edge/tests/test_video_stream.py

"""
视频流模块单元测试
"""

import unittest
from unittest.mock import MagicMock, patch, PropertyMock
from datetime import datetime
import sys
import os
import time

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))


class TestVideoFrame(unittest.TestCase):
    """测试视频帧类"""
    
    def test_video_frame_creation(self):
        """测试视频帧创建"""
        from core.video_stream import VideoFrame
        import numpy as np
        
        frame = VideoFrame(
            frame_id="frame001",
            camera_id="cam001",
            image=np.zeros((480, 640, 3), dtype=np.uint8),
            timestamp=datetime.now(),
            frame_number=1,
            width=640,
            height=480,
            fps=5.0
        )
        
        self.assertEqual(frame.frame_id, "frame001")
        self.assertEqual(frame.camera_id, "cam001")
        self.assertEqual(frame.width, 640)
        self.assertEqual(frame.height, 480)
    
    def test_video_frame_to_dict(self):
        """测试转换为字典"""
        from core.video_stream import VideoFrame
        import numpy as np
        
        frame = VideoFrame(
            frame_id="frame001",
            camera_id="cam001",
            image=np.zeros((480, 640, 3), dtype=np.uint8),
            timestamp=datetime.now(),
            frame_number=1,
            width=640,
            height=480,
            fps=5.0
        )
        
        result = frame.to_dict()
        
        self.assertEqual(result["frame_id"], "frame001")
        self.assertEqual(result["camera_id"], "cam001")
        self.assertEqual(result["frame_number"], 1)


class TestRTSPStreamReader(unittest.TestCase):
    """测试RTSP流读取器"""
    
    def test_stream_reader_creation(self):
        """测试读取器创建"""
        from core.video_stream import RTSPStreamReader
        
        reader = RTSPStreamReader(
            camera_id="cam001",
            rtsp_url="rtsp://192.168.1.1:554/stream1",
            target_fps=5
        )
        
        self.assertEqual(reader.camera_id, "cam001")
        self.assertEqual(reader.target_fps, 5)
        self.assertFalse(reader.is_connected)
        self.assertFalse(reader.is_running)
    
    def test_stream_status(self):
        """测试流状态"""
        from core.video_stream import StreamStatus
        
        status = StreamStatus(
            camera_id="cam001",
            is_connected=True,
            is_running=True,
            last_frame_time=datetime.now(),
            frame_count=100,
            reconnect_attempts=0,
            error_message=None
        )
        
        self.assertEqual(status.camera_id, "cam001")
        self.assertTrue(status.is_connected)
        self.assertTrue(status.is_running)
        self.assertEqual(status.frame_count, 100)


class TestMultiStreamManager(unittest.TestCase):
    """测试多流管理器"""
    
    def test_manager_creation(self):
        """测试管理器创建"""
        from core.video_stream import MultiStreamManager
        
        manager = MultiStreamManager()
        
        self.assertEqual(len(manager.get_all_streams()), 0)
    
    def test_add_remove_stream(self):
        """测试添加和移除流"""
        from core.video_stream import MultiStreamManager
        
        manager = MultiStreamManager()
        
        reader = manager.add_stream(
            camera_id="cam001",
            rtsp_url="rtsp://192.168.1.1:554/stream1"
        )
        
        self.assertEqual(len(manager.get_all_streams()), 1)
        
        manager.remove_stream("cam001")
        
        self.assertEqual(len(manager.get_all_streams()), 0)
    
    def test_get_statistics(self):
        """测试获取统计"""
        from core.video_stream import MultiStreamManager
        
        manager = MultiStreamManager()
        manager.add_stream(
            camera_id="cam001",
            rtsp_url="rtsp://192.168.1.1:554/stream1"
        )
        
        stats = manager.get_statistics()
        
        self.assertEqual(stats["total_streams"], 1)
        self.assertIn("streams", stats)


if __name__ == "__main__":
    unittest.main()