84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""测试body配置功能"""
|
|
from tasks import daily_morning_task
|
|
from generic_api_task import generic_api_call
|
|
import api_config
|
|
import time
|
|
import json
|
|
|
|
def test_body_configuration():
|
|
"""测试不同API配置的body参数"""
|
|
print("=== 测试Body配置功能 ===")
|
|
|
|
# 测试不同的API配置
|
|
test_apis = ['main_api', 'backup_api', 'get_api', 'put_api']
|
|
|
|
for api_key in test_apis:
|
|
print(f"\n--- 测试 {api_key} 配置 ---")
|
|
|
|
# 显示配置信息
|
|
config = api_config.API_CONFIG[api_key]
|
|
print(f"URL: {config['url']}")
|
|
print(f"Method: {config['method']}")
|
|
print(f"Body: {json.dumps(config.get('body', {}), indent=2, ensure_ascii=False)}")
|
|
|
|
# 执行任务
|
|
result = generic_api_call.delay(api_key)
|
|
print(f"任务ID: {result.id}")
|
|
|
|
# 等待完成
|
|
for i in range(10):
|
|
try:
|
|
if result.ready():
|
|
print(f"任务状态: {result.status}")
|
|
if result.successful():
|
|
task_result = result.get()
|
|
print(f"✅ 任务成功: {task_result.get('message')}")
|
|
print(f"发送的Body: {json.dumps(api_config.get('body', {}), indent=2, ensure_ascii=False)}")
|
|
else:
|
|
print(f"❌ 任务失败: {result.result}")
|
|
break
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
print(f"检查任务状态时出错: {e}")
|
|
time.sleep(1)
|
|
|
|
time.sleep(2) # 等待一下再进行下一个测试
|
|
|
|
print("\n✅ 所有body配置测试完成")
|
|
|
|
def test_custom_body():
|
|
"""测试自定义body参数"""
|
|
print("\n=== 测试自定义Body参数 ===")
|
|
|
|
custom_payload = {
|
|
'title': '自定义任务',
|
|
'body': '这是自定义的请求体数据',
|
|
'userId': 999,
|
|
'extra_field': '额外字段'
|
|
}
|
|
|
|
print(f"自定义Body: {json.dumps(custom_payload, indent=2, ensure_ascii=False)}")
|
|
|
|
result = generic_api_call.delay('main_api', custom_payload)
|
|
print(f"任务ID: {result.id}")
|
|
|
|
# 等待完成
|
|
for i in range(10):
|
|
try:
|
|
if result.ready():
|
|
print(f"任务状态: {result.status}")
|
|
if result.successful():
|
|
task_result = result.get()
|
|
print(f"✅ 自定义任务成功: {task_result.get('message')}")
|
|
else:
|
|
print(f"❌ 自定义任务失败: {result.result}")
|
|
break
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
print(f"检查任务状态时出错: {e}")
|
|
time.sleep(1)
|
|
|
|
if __name__ == '__main__':
|
|
test_body_configuration()
|
|
test_custom_body() |