import json import time import unittest import os import sys # 添加必要的路径以便导入模块 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from core.t4_daemon import _manager from base.spider import BaseSpider from cachetools import cached, TTLCache # 每次测试完自动清理可能存在的类共享变量问题(0关闭 1启用) AUTO_CLEAR = 0 # 计算斐波那契数列 def get_cache_key(n): return n # 不加缓存的递归实现 def fibonacci_no_cache(n): if n < 2: return n return fibonacci_no_cache(n - 1) + fibonacci_no_cache(n - 2) # 加缓存的递归实现 @cached(cache=TTLCache(maxsize=100, ttl=3600), key=get_cache_key) def fibonacci_with_cache(n): if n < 2: return n return fibonacci_with_cache(n - 1) + fibonacci_with_cache(n - 2) def to_json(item): if not item: return item if isinstance(item, dict) or isinstance(item, list): return json.dumps(item, ensure_ascii=False) return item def to_object(item): if not item: return item if isinstance(item, dict) or isinstance(item, list): return item try: # 尝试解析JSON字符串 return json.loads(item) except (json.JSONDecodeError, TypeError): # 如果不是有效的JSON字符串，返回原值 return item class TestDemoFunctions(unittest.TestCase): def test_get_cache_key(self): self.assertEqual(get_cache_key(5), 5) self.assertEqual(get_cache_key("test"), "test") def test_fibonacci_no_cache(self): self.assertEqual(fibonacci_no_cache(0), 0) self.assertEqual(fibonacci_no_cache(1), 1) self.assertEqual(fibonacci_no_cache(5), 5) self.assertEqual(fibonacci_no_cache(10), 55) def test_fibonacci_with_cache(self): # 清除缓存以确保测试准确性 fibonacci_with_cache.cache_clear() self.assertEqual(fibonacci_with_cache(0), 0) self.assertEqual(fibonacci_with_cache(1), 1) self.assertEqual(fibonacci_with_cache(5), 5) self.assertEqual(fibonacci_with_cache(10), 55) # 测试缓存是否工作 start_time = time.time() result1 = fibonacci_with_cache(10) first_call_time = time.time() - start_time start_time = time.time() result2 = fibonacci_with_cache(10) second_call_time = time.time() - start_time self.assertEqual(result1, result2) self.assertLessEqual(second_call_time, first_call_time) def test_to_json(self): # 测试字典转换 test_dict = {"key": "value", "num": 123} result = to_json(test_dict) self.assertEqual(result, '{"key": "value", "num": 123}') # 测试列表转换 test_list = [1, 2, 3] result = to_json(test_list) self.assertEqual(result, '[1, 2, 3]') # 测试字符串不变 test_str = "already a string" result = to_json(test_str) self.assertEqual(result, test_str) # 测试None不变 self.assertIsNone(to_json(None)) def test_to_object(self): # 测试JSON字符串转换 json_str = '{"key": "value", "num": 123}' result = to_object(json_str) self.assertEqual(result, {"key": "value", "num": 123}) # 测试字典不变 test_dict = {"key": "value"} result = to_object(test_dict) self.assertEqual(result, test_dict) # 测试列表不变 test_list = [1, 2, 3] result = to_object(test_list) self.assertEqual(result, test_list) # 测试非JSON字符串返回原值 test_str = "not a json string" result = to_object(test_str) self.assertEqual(result, test_str) # 测试None不变 self.assertIsNone(to_object(None)) # 测试无效JSON字符串返回原值 invalid_json = "{invalid: json}" result = to_object(invalid_json) self.assertEqual(result, invalid_json) def test_encrypt_demo(self): input_str = '这是需要gzip加密的字符串' output_str = BaseSpider.gzip(input_str) self.assertIsInstance(output_str, str) decompressed_str = BaseSpider.ungzip(output_str) self.assertEqual(decompressed_str, input_str) def test_speed_demo(self): n = 25 # 测试无缓存版本 t1 = time.time() result1 = fibonacci_no_cache(n) t2 = time.time() cost_no_cache = round((t2 - t1) * 1000, 8) # 测试有缓存版本 t3 = time.time() result2 = fibonacci_with_cache(n) t4 = time.time() cost_with_cache = round((t4 - t3) * 1000, 8) # 结果应该相同 self.assertEqual(result1, result2) # 有缓存的版本应该更快 self.assertLess(cost_with_cache, cost_no_cache * 10) print(f'无缓存耗时: {cost_no_cache:.6f}毫秒, 有缓存耗时: {cost_with_cache:.6f}毫秒') def _test_script_functionality(self, script_path, ext_dict): """测试脚本功能的通用方法""" if not os.path.exists(script_path): self.skipTest(f"需要{script_path}文件才能运行此测试") env = { 'proxyUrl': '', 'ext': ext_dict, } env_str = to_json(env) # 测试home调用 home_result = _manager.call(script_path, 'home', env_str, [1]) home_result_obj = to_object(home_result) self.assertIsInstance(home_result_obj, dict) # 测试homeVod调用 homeVod_result = _manager.call(script_path, 'homeVod', env_str, []) homeVod_result_obj = to_object(homeVod_result) self.assertIsInstance(homeVod_result_obj, dict) # 如果有分类数据，继续测试其他功能 if home_result_obj and 'class' in home_result_obj and home_result_obj['class']: type_name = home_result_obj['class'][0]['type_name'] type_id = home_result_obj['class'][0]['type_id'] # 测试category调用 category_result = _manager.call(script_path, 'category', env_str, [type_id, 1, 1, {}]) category_result_obj = to_object(category_result) self.assertIsInstance(category_result_obj, dict) # 如果有分类列表数据，继续测试详情功能 if category_result_obj and 'list' in category_result_obj and category_result_obj['list']: vod_id = category_result_obj['list'][0]['vod_id'] # 测试detail调用 detail_result = _manager.call(script_path, 'detail', env_str, [[vod_id]]) detail_result_obj = to_object(detail_result) self.assertIsInstance(detail_result_obj, dict) # 如果有详情数据，继续测试播放功能 if detail_result_obj and 'list' in detail_result_obj and detail_result_obj['list']: vod_play_url = detail_result_obj['list'][0].get('vod_play_url', '') vod_play_from = detail_result_obj['list'][0].get('vod_play_from', '') if vod_play_url and vod_play_from: # 解析播放URL play_parts = vod_play_url.split('#')[0].split('$') if len(play_parts) > 1: play = play_parts[1] # 测试play调用 play_result = _manager.call(script_path, 'play', env_str, [vod_play_from, play, []]) play_result_obj = to_object(play_result) self.assertIsInstance(play_result_obj, dict) # 测试search调用 key = '测试' search_result = _manager.call(script_path, 'search', env_str, [key, 0, 1]) search_result_obj = to_object(search_result) self.assertIsInstance(search_result_obj, dict) return True def test_apphs_script_shiguang(self): """测试AppHs.py脚本，使用shiguang配置""" script_path = './AppHs.py' ext_dict = { "host": "https://dy.jmzp.net.cn", "app_id": "shiguang", "deviceid": "", "versionCode": "10000", "UMENG_CHANNEL": "guan" } self._test_script_functionality(script_path, ext_dict) def test_apphs_script_xuebao(self): """测试AppHs.py脚本，使用xuebao配置""" script_path = './AppHs.py' ext_dict = { "host": "https://dy.jszdzs.com", "app_id": "xuebao", "deviceid": "", "versionCode": "21300", "UMENG_CHANNEL": "share" } self._test_script_functionality(script_path, ext_dict) def test_apphs_script_haigou(self): """测试AppHs.py脚本，使用haigou配置""" script_path = './AppHs.py' ext_dict = { "host": "https://dy.stxbed.com", "app_id": "haigou", "deviceid": "", "versionCode": "20100", "UMENG_CHANNEL": "zhuan" } self.functionality = self._test_script_functionality(script_path, ext_dict) def test_appyqk_script_1(self): """测试AppYqk.py脚本，使用一起看配置""" script_path = './AppYqk.py' ext_dict = { "host": "https://gapi0320.3njzmrx1.com/config.json,https://gapi0320.lq0okex8.com/config.json,https://gapi0320.zabqs8xp.com/config.json,https://yappconfig-20250628-1318635097.cos.ap-shanghai.myqcloud.com/config.json,https://yconfig-20250628-1360051343.cos.ap-guangzhou.myqcloud.com/config.json", "appId": "d6d520ea90904f1ba680ed6c9c9f9007", "appkey": "70af67d2b6cf47679b397ea4c1886877", "udid": "bfc18c00-c866-46cb-8d7b-121c39b942d4", "bundlerId": "com.flotimingo.ts", "source": "1001_default", "version": "1.3.10", "versionCode": 1104} self._test_script_functionality(script_path, ext_dict) def test_apphs_script_haigou2(self): from AppHs import Spider spider = Spider(t4_api='') ext_dict = { "host": "https://dy.stxbed.com", "app_id": "haigou", "deviceid": "", "versionCode": "20100", "UMENG_CHANNEL": "zhuan" } spider.setExtendInfo(to_json(ext_dict)) spider.init('') home_result = spider.homeContent(1) print(home_result) self.assertIsInstance(home_result, dict) home_video_result = spider.homeVideoContent() print(home_video_result) self.assertIsInstance(home_video_result, dict) def test_qimao_novel_script(self): """测试七猫小说脚本""" script_path = './七猫小说[书].py' ext_dict = {} # 根据需要添加七猫小说的配置 self._test_script_functionality(script_path, ext_dict) def tearDown(self): """在每个测试方法后清理可能的状态""" if AUTO_CLEAR: # 清理管理器中的实例和正在初始化的对象 with _manager._lock: _manager._instances.clear() _manager._inflight.clear() # 清理可能存在的模块引用 for module_name in list(sys.modules.keys()): if module_name.startswith("t4_spider_"): del sys.modules[module_name] if __name__ == '__main__': unittest.main()