show_demo/test_sync.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
sync_play_video函数单元测试
测试KodiServer类中的sync_play_video方法的各种场景
从配置文件中读取客户端配置
"""

import sys
import os
import unittest
from unittest.mock import Mock, patch, MagicMock
import time
import yaml

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

from kodi_util.kodi_server import KodiServer
from kodi_util.kodi_module import KodiClient


class TestSyncPlayVideoWithConfig(unittest.TestCase):
    """测试sync_play_video函数的单元测试类 - 从配置文件读取配置"""
    
    def setUp(self):
        """测试前的准备工作"""
        # 读取配置文件
        self.config_path = "config/config.yaml"
        self.sound_config_path = "config/ButtonListenerTask.yaml"
        
        # 加载主配置
        with open(self.config_path, 'r', encoding='utf-8') as f:
            self.main_config = yaml.safe_load(f)
        
        # 加载声音配置
        with open(self.sound_config_path, 'r', encoding='utf-8') as f:
            self.sound_config = yaml.safe_load(f)
        
        # 获取客户端配置
        self.client_configs = self.main_config.get('kodi_clients', [])
        self.sound_client_index = self.sound_config.get('sound', {}).get('client_index', 0)
        self.default_volume = self.sound_config.get('sound', {}).get('volume', 100)
        
        # 创建mock客户端，基于配置文件中的客户端数量
        self.mock_clients = []
        for i, client_config in enumerate(self.client_configs):
            mock_client = Mock(spec=KodiClient)
            mock_client.host = client_config.get('ip', f'192.168.1.{100+i}')
            mock_client.port = client_config.get('port', 8080)
            mock_client.username = client_config.get('username', 'kodi')
            mock_client.password = client_config.get('password', '123')
            mock_client.stop_playback.return_value = True
            mock_client.set_volume.return_value = True
            mock_client.play_video.return_value = {"result": "OK"}
            self.mock_clients.append(mock_client)
        
        # 创建KodiServer实例并手动设置客户端
        with patch('kodi_util.kodi_server.KodiStateMonitorThread'):
            with patch('os.path.exists', return_value=False):
                # 跳过配置文件加载，直接创建空的KodiServer
                self.kodi_server = KodiServer.__new__(KodiServer)
                self.kodi_server.clients = []
                self.kodi_server.state_monitor = None
    
    def test_sync_play_video_no_clients(self):
        """测试没有客户端时的情况"""
        result = self.kodi_server.sync_play_video("/path/to/video.mp4")
        
        self.assertFalse(result["success"])
        self.assertEqual(result["message"], "没有可用的客户端")
    
    def test_sync_play_video_with_config_clients_success(self):
        """测试使用配置文件中的客户端成功播放的情况"""
        # 设置客户端（使用配置文件中的前3个客户端进行测试）
        test_clients = self.mock_clients[:3] if len(self.mock_clients) >= 3 else self.mock_clients
        self.kodi_server.clients = test_clients
        
        with patch('time.sleep'):  # Mock sleep以加速测试
            result = self.kodi_server.sync_play_video(
                "/sdcard/Movies/0.mp4",  # 使用配置文件中的视频路径
                sound_client_index=self.sound_client_index,
                default_volume=self.default_volume
            )
        
        # 验证结果
        self.assertTrue(result["success"])
        self.assertEqual(len(result["success_clients"]), len(test_clients))
        self.assertEqual(len(result["failed_clients"]), 0)
        
        # 验证所有客户端的stop_playback被调用
        for client in test_clients:
            client.stop_playback.assert_called()
            client.play_video.assert_called_with("/sdcard/Movies/0.mp4", loop=False)
    
    def test_sync_play_video_with_config_sound_settings(self):
        """测试使用配置文件中的声音设置"""
        # 使用配置文件中的前2个客户端
        test_clients = self.mock_clients[:2] if len(self.mock_clients) >= 2 else self.mock_clients
        self.kodi_server.clients = test_clients
        
        with patch('time.sleep'):
            with patch('builtins.print') as mock_print:
                result = self.kodi_server.sync_play_video(
                    "/sdcard/Movies/硫酸钠.mp4",  # 使用配置文件中的视频路径
                    sound_client_index=self.sound_client_index,
                    default_volume=self.default_volume
                )
        
        # 验证结果
        self.assertTrue(result["success"])
        
        # 验证音量设置根据配置文件
        if self.sound_client_index == -1:
            # 所有客户端都播放声音
            for client in test_clients:
                client.set_volume.assert_called_with(self.default_volume)
        else:
            # 只有指定客户端播放声音
            for idx, client in enumerate(test_clients):
                if idx == self.sound_client_index:
                    client.set_volume.assert_called_with(self.default_volume)
                    mock_print.assert_any_call(f"客户端 {client.host} 将播放声音，音量设置为 {self.default_volume}%")
                else:
                    client.set_volume.assert_called_with(0)
                    mock_print.assert_any_call(f"客户端 {client.host} 将静音播放")
    
    def test_sync_play_video_all_config_videos(self):
        """测试配置文件中的所有视频路径"""
        if not self.mock_clients:
            self.skipTest("配置文件中没有客户端配置")
        
        # 使用第一个客户端进行测试
        self.kodi_server.clients = [self.mock_clients[0]]
        
        # 获取配置文件中的视频路径
        video_paths = self.main_config.get('video_paths', [])
        if not video_paths:
            video_paths = self.sound_config.get('video_paths', [])
        
        if not video_paths:
            self.skipTest("配置文件中没有视频路径配置")
        
        # 测试每个视频路径
        for video_path in video_paths[:3]:  # 只测试前3个视频路径
            with self.subTest(video_path=video_path):
                with patch('time.sleep'):
                    result = self.kodi_server.sync_play_video(
                        video_path,
                        sound_client_index=0,
                        default_volume=self.default_volume
                    )
                
                self.assertTrue(result["success"])
                self.mock_clients[0].play_video.assert_called_with(video_path, loop=False)
    
    def test_sync_play_video_invalid_sound_client_index_with_config(self):
        """测试无效的声音客户端索引（超出配置文件中的客户端数量）"""
        # 使用配置文件中的客户端
        test_clients = self.mock_clients[:2] if len(self.mock_clients) >= 2 else self.mock_clients
        self.kodi_server.clients = test_clients
        
        # 使用超出范围的索引
        invalid_index = len(test_clients) + 5
        
        with patch('time.sleep'):
            with patch('builtins.print') as mock_print:
                result = self.kodi_server.sync_play_video(
                    "/sdcard/Movies/0.mp4",
                    sound_client_index=invalid_index,
                    default_volume=self.default_volume
                )
        
        # 验证结果仍然成功（使用默认客户端）
        self.assertTrue(result["success"])
        
        # 验证打印了警告信息
        mock_print.assert_any_call(f"警告：指定的声音客户端索引 {invalid_index} 超出范围，将使用默认客户端")
    
    def test_sync_play_video_client_failure_with_config(self):
        """测试配置文件中的客户端播放失败的情况"""
        if len(self.mock_clients) < 2:
            self.skipTest("配置文件中客户端数量不足，需要至少2个客户端")
        
        # 设置第一个客户端播放失败，第二个客户端成功
        self.mock_clients[0].play_video.return_value = {"result": "ERROR"}
        self.mock_clients[1].play_video.return_value = {"result": "OK"}
        
        self.kodi_server.clients = self.mock_clients[:2]
        
        with patch('time.sleep'):
            result = self.kodi_server.sync_play_video(
                "/sdcard/Movies/0.mp4",
                sound_client_index=0,  # 第一个客户端是主客户端
                default_volume=self.default_volume
            )
        
        # 验证结果：部分成功
        self.assertTrue(result["success"])  # 至少有一个客户端成功
        self.assertEqual(len(result["success_clients"]), 1)
        self.assertEqual(len(result["failed_clients"]), 1)
        self.assertIn(self.mock_clients[1].host, result["success_clients"])
        self.assertIn(self.mock_clients[0].host, result["failed_clients"])
    
    def test_config_file_loading(self):
        """测试配置文件加载是否正确"""
        # 验证主配置文件加载
        self.assertIn('kodi_clients', self.main_config)
        self.assertIsInstance(self.main_config['kodi_clients'], list)
        self.assertGreater(len(self.main_config['kodi_clients']), 0)
        
        # 验证声音配置文件加载
        self.assertIn('sound', self.sound_config)
        self.assertIn('client_index', self.sound_config['sound'])
        self.assertIn('volume', self.sound_config['sound'])
        
        # 验证配置值的合理性
        self.assertIsInstance(self.sound_client_index, int)
        self.assertGreaterEqual(self.sound_client_index, -1)
        self.assertIsInstance(self.default_volume, int)
        self.assertGreaterEqual(self.default_volume, 0)
        self.assertLessEqual(self.default_volume, 100)
    
    def test_sync_play_video_with_real_config_integration(self):
        """集成测试：使用真实配置文件的完整测试"""
        if not self.mock_clients:
            self.skipTest("配置文件中没有客户端配置")
        
        # 使用配置文件中的所有客户端（最多5个以避免测试时间过长）
        test_clients = self.mock_clients[:5]
        self.kodi_server.clients = test_clients
        
        # 使用配置文件中的视频路径
        video_paths = self.main_config.get('video_paths', [])
        if not video_paths:
            video_paths = self.sound_config.get('video_paths', [])
        
        if video_paths:
            test_video = video_paths[0]
        else:
            test_video = "/sdcard/Movies/0.mp4"
        
        with patch('time.sleep'):
            with patch('builtins.print') as mock_print:
                result = self.kodi_server.sync_play_video(
                    test_video,
                    sound_client_index=self.sound_client_index,
                    default_volume=self.default_volume
                )
        
        # 验证结果
        self.assertTrue(result["success"])
        self.assertEqual(len(result["success_clients"]), len(test_clients))
        self.assertEqual(len(result["failed_clients"]), 0)
        
        # 验证所有客户端都被正确调用
        for client in test_clients:
            client.stop_playback.assert_called()
            client.play_video.assert_called_with(test_video, loop=False)
        
        print(f"\n集成测试完成:")
        print(f"- 测试客户端数量: {len(test_clients)}")
        print(f"- 声音客户端索引: {self.sound_client_index}")
        print(f"- 默认音量: {self.default_volume}")
        print(f"- 测试视频: {test_video}")


def run_tests():
    """运行所有测试"""
    print("开始运行sync_play_video函数单元测试（从配置文件读取配置）...")
    print("=" * 70)
    
    # 检查配置文件是否存在
    config_files = ["config/config.yaml", "config/ButtonListenerTask.yaml"]
    for config_file in config_files:
        if not os.path.exists(config_file):
            print(f"错误：配置文件 {config_file} 不存在")
            return False
    
    # 创建测试套件
    test_suite = unittest.TestLoader().loadTestsFromTestCase(TestSyncPlayVideoWithConfig)
    
    # 运行测试
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(test_suite)
    
    print("=" * 70)
    print(f"测试完成！")
    print(f"运行测试数量: {result.testsRun}")
    print(f"失败数量: {len(result.failures)}")
    print(f"错误数量: {len(result.errors)}")
    
    if result.failures:
        print("\n失败的测试:")
        for test, traceback in result.failures:
            print(f"- {test}: {traceback}")
    
    if result.errors:
        print("\n错误的测试:")
        for test, traceback in result.errors:
            print(f"- {test}: {traceback}")
    
    return result.wasSuccessful()


if __name__ == "__main__":
    success = run_tests()
    sys.exit(0 if success else 1)