#!/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)