7043347417
为了准确定位 file_path 为 null 的问题，创建了 test_end_to_end.py 测试脚本。 测试流程： 1. 创建测试配置 2. 调用 generate_daily_report 生成日报 3. 查询历史记录 4. 检查 file_path 是否正确保存 5. 清理测试数据 运行方式： ```bash python test_end_to_end.py ``` 这个测试会模拟完整的生成流程，帮助定位问题出在哪个环节。 Coze-Commit-Type: user Coze-User-ID: 3722323274763196 Coze-Conversation-ID: 9894087
112 lines
3.9 KiB
Python
112 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
完整的端到端测试：生成日报并检查历史记录
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from lib.report_generator import generate_daily_report
|
||
from lib.db import execute_query, execute_update
|
||
|
||
def test_full_flow():
|
||
"""测试完整流程"""
|
||
print("="*60)
|
||
print("完整端到端测试")
|
||
print("="*60)
|
||
|
||
try:
|
||
# 1. 创建测试配置
|
||
print("\n1. 创建测试配置...")
|
||
from lib.db import execute_insert
|
||
import json
|
||
|
||
config_id = 999998
|
||
config_name = "测试配置_端到端"
|
||
split_type = "company_id"
|
||
split_value = "test_company"
|
||
split_name = "测试公司"
|
||
selected_fields = ["order_no", "report_time", "total_money"]
|
||
sum_fields = ["total_money"]
|
||
|
||
# 删除旧配置
|
||
execute_update("DELETE FROM t_daily_report_config WHERE id = %s", (config_id,))
|
||
|
||
# 插入新配置
|
||
insert_sql = """
|
||
INSERT INTO t_daily_report_config
|
||
(id, config_name, split_type, split_value, split_name, selected_fields, sum_fields, is_active, sort_order)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
|
||
"""
|
||
execute_insert(insert_sql, (
|
||
config_id,
|
||
config_name,
|
||
split_type,
|
||
split_value,
|
||
split_name,
|
||
json.dumps(selected_fields, ensure_ascii=False),
|
||
json.dumps(sum_fields, ensure_ascii=False),
|
||
1,
|
||
0
|
||
))
|
||
print(f" ✓ 配置已创建，ID: {config_id}")
|
||
|
||
# 2. 生成日报
|
||
print("\n2. 生成日报...")
|
||
result = generate_daily_report(config_id)
|
||
|
||
print(f" 生成结果:")
|
||
print(f" - success: {result.get('success')}")
|
||
print(f" - file_path: {result.get('file_path')}")
|
||
print(f" - total_orders: {result.get('total_orders')}")
|
||
print(f" - history_id: {result.get('history_id')}")
|
||
|
||
if not result.get('success'):
|
||
print(f" ✗ 生成失败: {result.get('message')}")
|
||
return
|
||
|
||
# 3. 查询历史记录
|
||
print("\n3. 查询历史记录...")
|
||
history_id = result.get('history_id')
|
||
query_sql = "SELECT * FROM t_daily_report_history WHERE id = %s"
|
||
history_records = execute_query(query_sql, (history_id,))
|
||
|
||
if history_records and len(history_records) > 0:
|
||
record = history_records[0]
|
||
print(f" ✓ 历史记录查询成功")
|
||
print(f" - ID: {record.get('id')}")
|
||
print(f" - config_name: {record.get('config_name')}")
|
||
print(f" - file_path: '{record.get('file_path')}'")
|
||
print(f" - status: {record.get('status')}")
|
||
print(f" - total_orders: {record.get('total_orders')}")
|
||
|
||
if record.get('file_path') and record.get('file_path') != '':
|
||
print(f"\n✓✓✓ 端到端测试成功！file_path 已正确保存")
|
||
print(f" 文件路径: {record.get('file_path')}")
|
||
else:
|
||
print(f"\n❌❌❌ file_path 为空！")
|
||
else:
|
||
print(" ❌ 未找到历史记录")
|
||
|
||
# 4. 清理测试数据
|
||
print("\n4. 清理测试数据...")
|
||
execute_update("DELETE FROM t_daily_report_history WHERE config_id = %s", (config_id,))
|
||
execute_update("DELETE FROM t_daily_report_config WHERE id = %s", (config_id,))
|
||
print(" ✓ 已清理")
|
||
|
||
print("\n" + "="*60)
|
||
print("测试完成")
|
||
print("="*60)
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
input("\n按回车键退出...")
|
||
|
||
if __name__ == "__main__":
|
||
test_full_flow()
|