# AIfeng/2025-07-07 17:11:09
# 流式语音识别演示应用测试
# 测试StreamingRecognitionDemo类的核心功能和集成能力

import sys
import os
import json
import pytest
import tempfile
from unittest.mock import Mock, patch, MagicMock
from typing import Dict, Any

# 添加项目根目录到路径
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from streaming_demo import StreamingRecognitionDemo
from streaming.optimization.optimization_manager import OptimizationMode

class TestStreamingRecognitionDemo:
    """流式语音识别演示应用测试类"""
    
    def setup_method(self):
        """测试前置设置"""
        self.test_config = {
            "streaming_vad": {
                "sample_rate": 16000,
                "chunk_size": 1024,
                "volume_threshold": 0.03,
                "silence_duration": 1.0,
                "min_speech_duration": 0.3,
                "max_speech_duration": 15.0,
                "partial_result_interval": 2.0
            },
            "streaming_recognition": {
                "confidence_threshold": 0.6,
                "max_session_duration": 30.0,
                "result_merge_window": 1.0
            },
            "streaming_recorder": {
                "audio": {
                    "rate": 16000,
                    "chunk": 1024,
                    "channels": 1
                }
            }
        }
    
    def test_demo_initialization_with_valid_config(self):
        """测试使用有效配置初始化演示应用"""
        # 创建临时配置文件
        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
            json.dump(self.test_config, f)
            config_path = f.name
        
        try:
            demo = StreamingRecognitionDemo(config_path)
            
            # 验证初始化状态
            assert demo.config == self.test_config
            assert demo.recorder is None
            assert demo.asr_client is None
            assert demo.is_running is False
            assert demo.selected_device_index is None
            assert demo.session_results == {}
            assert demo.current_optimization_mode == OptimizationMode.BALANCED
            
            # 验证统计信息初始化
            expected_stats = {
                'sessions_created': 0,
                'partial_results': 0,
                'final_results': 0,
                'total_speech_duration': 0.0,
                'start_time': None,
                'optimization_metrics': {}
            }
            assert demo.stats == expected_stats
            
        finally:
            os.unlink(config_path)
    
    def test_demo_initialization_with_missing_config(self):
        """测试配置文件不存在时的初始化"""
        demo = StreamingRecognitionDemo("non_existent_config.json")
        
        # 应该使用默认配置
        assert demo.config is not None
        assert "streaming_vad" in demo.config
        assert "streaming_recognition" in demo.config
        assert "streaming_recorder" in demo.config
        
        # 验证默认配置的关键参数
        vad_config = demo.config["streaming_vad"]
        assert vad_config["sample_rate"] == 16000
        assert vad_config["volume_threshold"] == 0.03
    
    def test_demo_initialization_with_invalid_json(self):
        """测试无效JSON配置文件的处理"""
        # 创建无效JSON文件
        with tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as f:
            f.write("invalid json content {")
            config_path = f.name
        
        try:
            demo = StreamingRecognitionDemo(config_path)
            
            # 应该回退到默认配置
            assert demo.config is not None
            assert "streaming_vad" in demo.config
            
        finally:
            os.unlink(config_path)
    
    @patch('streaming_demo.FunASRSync')
    def test_setup_asr_client_success(self, mock_funasr):
        """测试ASR客户端设置成功"""
        mock_asr_instance = Mock()
        mock_funasr.return_value = mock_asr_instance
        
        demo = StreamingRecognitionDemo()
        result = demo._setup_asr_client()
        
        assert result is True
        assert demo.asr_client == mock_asr_instance
        mock_funasr.assert_called_once_with(username="streaming_demo")
    
    @patch('streaming_demo.FunASRSync')
    def test_setup_asr_client_failure(self, mock_funasr):
        """测试ASR客户端设置失败"""
        mock_funasr.side_effect = Exception("ASR initialization failed")
        
        demo = StreamingRecognitionDemo()
        result = demo._setup_asr_client()
        
        assert result is False
        assert demo.asr_client is None
    
    @patch('streaming_demo.StreamingRecorder')
    def test_setup_recorder_success(self, mock_recorder_class):
        """测试录音器设置成功"""
        mock_recorder = Mock()
        mock_recorder_class.return_value = mock_recorder
        
        demo = StreamingRecognitionDemo()
        result = demo._setup_recorder()
        
        assert result is True
        assert demo.recorder == mock_recorder
        
        # 验证录音器初始化参数
        mock_recorder_class.assert_called_once()
        call_kwargs = mock_recorder_class.call_args[1]
        assert call_kwargs['chunk'] == 1024
        assert call_kwargs['rate'] == 16000
        assert call_kwargs['channels'] == 1
        assert call_kwargs['username'] == "streaming_demo"
        
        # 验证回调函数设置
        assert mock_recorder.on_partial_result == demo._on_partial_result
        assert mock_recorder.on_final_result == demo._on_final_result
        assert mock_recorder.on_session_complete == demo._on_session_complete
        assert mock_recorder.on_status_update == demo._on_status_update
    
    @patch('streaming_demo.StreamingRecorder')
    def test_setup_recorder_failure(self, mock_recorder_class):
        """测试录音器设置失败"""
        mock_recorder_class.side_effect = Exception("Recorder initialization failed")
        
        demo = StreamingRecognitionDemo()
        result = demo._setup_recorder()
        
        assert result is False
        assert demo.recorder is None
    
    def test_partial_result_callback(self):
        """测试部分识别结果回调"""
        demo = StreamingRecognitionDemo()
        
        session_id = "test_session_001"
        text = "测试文本"
        confidence = 0.85
        
        # 调用回调函数
        demo._on_partial_result(session_id, text, confidence)
        
        # 验证统计更新
        assert demo.stats['partial_results'] == 1
        
        # 验证会话结果记录
        assert session_id in demo.session_results
        session_data = demo.session_results[session_id]
        assert 'partial_results' in session_data
        assert 'final_result' in session_data
        assert 'start_time' in session_data
        
        # 验证部分结果记录
        partial_results = session_data['partial_results']
        assert len(partial_results) == 1
        assert partial_results[0]['text'] == text
        assert partial_results[0]['confidence'] == confidence
        assert 'timestamp' in partial_results[0]
    
    def test_final_result_callback(self):
        """测试最终识别结果回调"""
        demo = StreamingRecognitionDemo()
        
        session_id = "test_session_002"
        text = "最终测试文本"
        confidence = 0.92
        
        # 先创建会话（模拟部分结果）
        demo._on_partial_result(session_id, "部分文本", 0.8)
        
        # 调用最终结果回调
        demo._on_final_result(session_id, text, confidence)
        
        # 验证统计更新
        assert demo.stats['final_results'] == 1
        
        # 验证最终结果记录
        session_data = demo.session_results[session_id]
        final_result = session_data['final_result']
        assert final_result['text'] == text
        assert final_result['confidence'] == confidence
        assert 'timestamp' in final_result
    
    def test_session_complete_callback(self):
        """测试会话完成回调"""
        demo = StreamingRecognitionDemo()
        
        session_id = "test_session_003"
        final_text = "完整的会话文本"
        
        # 先创建会话
        demo._on_partial_result(session_id, "部分文本", 0.8)
        
        # 等待一小段时间确保时间差
        import time
        time.sleep(0.01)
        
        # 调用会话完成回调
        demo._on_session_complete(session_id, final_text)
        
        # 验证语音时长统计更新
        assert demo.stats['total_speech_duration'] > 0
    
    def test_status_update_callback_vad_status(self):
        """测试VAD状态更新回调"""
        demo = StreamingRecognitionDemo()
        
        status = {
            'type': 'vad_status',
            'is_speaking': True,
            'volume': 0.05
        }
        
        # 调用状态更新回调（不应该抛出异常）
        demo._on_status_update(status)
        
        # 验证_last_vad_update属性被设置
        assert hasattr(demo, '_last_vad_update')
    
    def test_status_update_callback_optimization_metrics(self):
        """测试优化指标状态更新回调"""
        demo = StreamingRecognitionDemo()
        
        session_id = "test_session_004"
        metrics = {
            'total_latency_ms': 150.5,
            'accuracy_score': 0.88
        }
        
        status = {
            'type': 'optimization_metrics',
            'session_id': session_id,
            'metrics': metrics
        }
        
        # 调用状态更新回调
        demo._on_status_update(status)
        
        # 验证优化指标记录
        assert session_id in demo.stats['optimization_metrics']
        assert demo.stats['optimization_metrics'][session_id] == metrics
    
    def test_status_update_callback_optimization_error(self):
        """测试优化错误状态更新回调"""
        demo = StreamingRecognitionDemo()
        
        status = {
            'type': 'optimization_error',
            'session_id': 'test_session_005',
            'error_type': 'processing_timeout',
            'message': 'Processing timeout occurred'
        }
        
        # 调用状态更新回调（不应该抛出异常）
        demo._on_status_update(status)
    
    def test_status_update_callback_mode_changed(self):
        """测试优化模式变更状态更新回调"""
        demo = StreamingRecognitionDemo()
        
        status = {
            'type': 'optimization_mode_changed',
            'mode': 'PERFORMANCE'
        }
        
        # 调用状态更新回调（不应该抛出异常）
        demo._on_status_update(status)
    
    @patch('streaming_demo.StreamingRecorder')
    def test_list_audio_devices(self, mock_recorder_class):
        """测试音频设备列表功能"""
        # 模拟设备列表
        mock_devices = [
            {'index': 0, 'name': 'Default Device', 'channels': 2, 'sample_rate': 44100},
            {'index': 1, 'name': 'USB Microphone', 'channels': 1, 'sample_rate': 16000}
        ]
        
        mock_recorder = Mock()
        mock_recorder.list_audio_devices.return_value = mock_devices
        mock_recorder_class.return_value = mock_recorder
        
        demo = StreamingRecognitionDemo()
        
        # 调用设备列表功能（不应该抛出异常）
        demo._list_audio_devices()
        
        # 验证临时录音器被创建和调用
        mock_recorder_class.assert_called_once()
        mock_recorder.list_audio_devices.assert_called_once()
    
    @patch('streaming_demo.StreamingRecorder')
    def test_list_audio_devices_failure(self, mock_recorder_class):
        """测试音频设备列表功能失败"""
        mock_recorder_class.side_effect = Exception("Device enumeration failed")
        
        demo = StreamingRecognitionDemo()
        
        # 调用设备列表功能（不应该抛出异常）
        demo._list_audio_devices()
    
    def test_switch_optimization_mode(self):
        """测试优化模式切换功能"""
        demo = StreamingRecognitionDemo()
        
        # 验证初始模式
        assert demo.current_optimization_mode == OptimizationMode.BALANCED
        
        # 模拟用户输入（这里只测试模式设置逻辑）
        new_mode = OptimizationMode.SPEED_FIRST
        demo.current_optimization_mode = new_mode
        
        assert demo.current_optimization_mode == OptimizationMode.SPEED_FIRST
    
    @patch('streaming_demo.StreamingRecorder')
    def test_switch_optimization_mode_with_recorder(self, mock_recorder_class):
        """测试有录音器时的优化模式切换"""
        mock_recorder = Mock()
        mock_optimization_manager = Mock()
        mock_recorder.optimization_manager = mock_optimization_manager
        mock_recorder_class.return_value = mock_recorder
        
        demo = StreamingRecognitionDemo()
        demo._setup_recorder()
        
        # 切换模式
        new_mode = OptimizationMode.ACCURACY_FIRST
        demo.current_optimization_mode = new_mode
        
        # 如果有录音器，应该调用优化管理器的设置方法
        if hasattr(demo.recorder, 'optimization_manager'):
            demo.recorder.optimization_manager.set_optimization_mode(new_mode)
            mock_optimization_manager.set_optimization_mode.assert_called_once_with(new_mode)
    
    def test_cleanup_without_components(self):
        """测试没有组件时的清理"""
        demo = StreamingRecognitionDemo()
        
        # 调用清理（不应该抛出异常）
        demo._cleanup()
        
        assert demo.is_running is False
    
    def test_cleanup_with_components(self):
        """测试有组件时的清理"""
        demo = StreamingRecognitionDemo()
        
        # 模拟组件
        mock_recorder = Mock()
        mock_recorder.is_recording.return_value = True
        demo.recorder = mock_recorder
        
        mock_asr_client = Mock()
        demo.asr_client = mock_asr_client
        
        # 调用清理
        demo._cleanup()
        
        # 验证清理调用
        mock_recorder.stop_recording.assert_called_once()
        mock_asr_client.end.assert_called_once()
        assert demo.is_running is False
    
    def test_cleanup_with_asr_client_error(self):
        """测试ASR客户端清理时出错"""
        demo = StreamingRecognitionDemo()
        
        # 模拟ASR客户端清理出错
        mock_asr_client = Mock()
        mock_asr_client.end.side_effect = Exception("Cleanup error")
        demo.asr_client = mock_asr_client
        
        # 调用清理（不应该抛出异常）
        demo._cleanup()
        
        assert demo.is_running is False
    
    def test_get_default_config(self):
        """测试默认配置获取"""
        demo = StreamingRecognitionDemo()
        default_config = demo._get_default_config()
        
        # 验证默认配置结构
        assert "streaming_vad" in default_config
        assert "streaming_recognition" in default_config
        assert "streaming_recorder" in default_config
        
        # 验证关键配置项
        vad_config = default_config["streaming_vad"]
        assert vad_config["sample_rate"] == 16000
        assert vad_config["chunk_size"] == 1024
        assert vad_config["volume_threshold"] == 0.03
        
        recognition_config = default_config["streaming_recognition"]
        assert recognition_config["confidence_threshold"] == 0.6
        assert recognition_config["max_session_duration"] == 30.0
        
        audio_config = default_config["streaming_recorder"]["audio"]
        assert audio_config["rate"] == 16000
        assert audio_config["chunk"] == 1024
        assert audio_config["channels"] == 1

class TestStreamingDemoIntegration:
    """流式语音识别演示应用集成测试"""
    
    @patch('streaming_demo.FunASRSync')
    @patch('streaming_demo.StreamingRecorder')
    def test_demo_initialization_integration(self, mock_recorder_class, mock_funasr):
        """测试演示应用完整初始化流程"""
        # 模拟成功的组件初始化
        mock_asr_instance = Mock()
        mock_funasr.return_value = mock_asr_instance
        
        mock_recorder = Mock()
        mock_recorder_class.return_value = mock_recorder
        
        demo = StreamingRecognitionDemo()
        
        # 测试ASR客户端设置
        asr_result = demo._setup_asr_client()
        assert asr_result is True
        assert demo.asr_client == mock_asr_instance
        
        # 测试录音器设置
        recorder_result = demo._setup_recorder()
        assert recorder_result is True
        assert demo.recorder == mock_recorder
        
        # 验证回调函数设置
        assert mock_recorder.on_partial_result == demo._on_partial_result
        assert mock_recorder.on_final_result == demo._on_final_result
        assert mock_recorder.on_session_complete == demo._on_session_complete
        assert mock_recorder.on_status_update == demo._on_status_update
    
    def test_demo_callback_workflow(self):
        """测试演示应用回调工作流程"""
        demo = StreamingRecognitionDemo()
        
        session_id = "integration_test_session"
        
        # 模拟完整的识别流程
        # 1. 部分结果
        demo._on_partial_result(session_id, "你好", 0.7)
        demo._on_partial_result(session_id, "你好世界", 0.8)
        
        # 2. 最终结果
        demo._on_final_result(session_id, "你好世界！", 0.9)
        
        # 等待一小段时间确保时间差
        import time
        time.sleep(0.01)
        
        # 3. 会话完成
        demo._on_session_complete(session_id, "你好世界！")
        
        # 验证统计信息
        assert demo.stats['partial_results'] == 2
        assert demo.stats['final_results'] == 1
        assert demo.stats['total_speech_duration'] > 0
        
        # 验证会话数据
        assert session_id in demo.session_results
        session_data = demo.session_results[session_id]
        assert len(session_data['partial_results']) == 2
        assert session_data['final_result']['text'] == "你好世界！"
        assert session_data['final_result']['confidence'] == 0.9
    
    def test_demo_optimization_metrics_workflow(self):
        """测试演示应用优化指标工作流程"""
        demo = StreamingRecognitionDemo()
        
        # 模拟多个会话的优化指标
        sessions_metrics = {
            "session_001": {
                'total_latency_ms': 120.5,
                'accuracy_score': 0.85
            },
            "session_002": {
                'total_latency_ms': 95.2,
                'accuracy_score': 0.92
            },
            "session_003": {
                'total_latency_ms': 180.1,
                'accuracy_score': 0.78
            }
        }
        
        # 模拟优化指标状态更新
        for session_id, metrics in sessions_metrics.items():
            status = {
                'type': 'optimization_metrics',
                'session_id': session_id,
                'metrics': metrics
            }
            demo._on_status_update(status)
        
        # 验证优化指标记录
        assert len(demo.stats['optimization_metrics']) == 3
        for session_id, expected_metrics in sessions_metrics.items():
            assert session_id in demo.stats['optimization_metrics']
            assert demo.stats['optimization_metrics'][session_id] == expected_metrics
        
        # 验证平均指标计算（模拟_print_status中的逻辑）
        total_latency = sum(m['total_latency_ms'] for m in sessions_metrics.values())
        total_accuracy = sum(m['accuracy_score'] for m in sessions_metrics.values())
        count = len(sessions_metrics)
        
        avg_latency = total_latency / count
        avg_accuracy = total_accuracy / count
        
        assert abs(avg_latency - 131.93) < 0.1  # 约131.93ms
        assert abs(avg_accuracy - 0.85) < 0.01  # 约0.85

if __name__ == "__main__":
    pytest.main([__file__, "-v"])