322 lines
12 KiB
Python
322 lines
12 KiB
Python
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()
|