193 lines
6.1 KiB
Python
193 lines
6.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
最终验证脚本：确保所有问题都已修复
|
|
1. 创建测试数据
|
|
2. 测试update_query_result接口
|
|
3. 验证数据是否正确保存
|
|
"""
|
|
|
|
import sqlite3
|
|
import requests
|
|
import json
|
|
import logging
|
|
|
|
# 配置日志
|
|
handlers = [logging.FileHandler('final_verification.log'), logging.StreamHandler()]
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s',
|
|
handlers=handlers
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# API地址
|
|
BASE_URL = 'http://localhost:99'
|
|
API_ENDPOINT = f'{BASE_URL}/api/update-query-result'
|
|
DB_PATH = 'car_info.db'
|
|
|
|
def create_test_data():
|
|
"""
|
|
在数据库中创建测试数据
|
|
"""
|
|
logger.info("开始创建测试数据")
|
|
|
|
try:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
|
|
# 创建一个测试查询历史记录
|
|
cursor.execute("""
|
|
INSERT INTO query_history (department, query_date, plate_number, phone, results_count, query_ip)
|
|
VALUES (?, datetime('now'), ?, ?, ?, ?)
|
|
""", ('测试部门', '湘AB1234', '13800138000', 0, '127.0.0.1'))
|
|
|
|
# 获取新插入的记录ID
|
|
query_id = cursor.lastrowid
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
logger.info(f"成功创建测试数据，生成的查询ID: {query_id}")
|
|
return query_id
|
|
|
|
except sqlite3.Error as e:
|
|
logger.error(f"创建测试数据时发生数据库错误: {str(e)}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"创建测试数据时发生错误: {str(e)}")
|
|
return None
|
|
|
|
def test_update_query_result_with_real_data(query_id):
|
|
"""
|
|
使用真实的测试数据测试update_query_result接口
|
|
"""
|
|
logger.info(f"使用查询ID {query_id} 测试update_query_result接口")
|
|
|
|
# 测试数据
|
|
test_data = {
|
|
'id': query_id,
|
|
'results_count': 2,
|
|
'cars_info': [
|
|
{
|
|
'plate_number': '湘AB1234',
|
|
'phone': '13800138000',
|
|
'name': '张三',
|
|
'email': 'zhangsan@example.com',
|
|
'address': '北京市海淀区'
|
|
},
|
|
{
|
|
'plate_number': '湘AB5678',
|
|
'phone': '13800138001',
|
|
'name': '李四',
|
|
'email': 'lisi@example.com',
|
|
'address': '上海市浦东新区'
|
|
}
|
|
]
|
|
}
|
|
|
|
try:
|
|
logger.info(f"发送测试请求，数据: {json.dumps(test_data, ensure_ascii=False)}")
|
|
|
|
# 发送POST请求
|
|
response = requests.post(
|
|
API_ENDPOINT,
|
|
json=test_data,
|
|
headers={'Content-Type': 'application/json'},
|
|
timeout=30
|
|
)
|
|
|
|
logger.info(f"请求完成，状态码: {response.status_code}")
|
|
|
|
# 输出响应内容
|
|
if response.status_code == 200:
|
|
try:
|
|
result = response.json()
|
|
logger.info(f"响应数据: {json.dumps(result, ensure_ascii=False)}")
|
|
return True, result
|
|
except json.JSONDecodeError:
|
|
logger.warning(f"响应不是有效的JSON格式: {response.text}")
|
|
return False, None
|
|
else:
|
|
logger.error(f"请求失败，状态码: {response.status_code}，响应内容: {response.text}")
|
|
return False, None
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
logger.error(f"请求异常: {str(e)}")
|
|
return False, None
|
|
|
|
def verify_data_in_db(query_id, car_owner_ids):
|
|
"""
|
|
验证数据是否正确保存到数据库
|
|
"""
|
|
logger.info(f"验证查询ID {query_id} 的数据是否正确保存到数据库")
|
|
|
|
try:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
|
|
# 检查query_history表是否更新成功
|
|
cursor.execute("SELECT results_count FROM query_history WHERE id = ?", (query_id,))
|
|
result = cursor.fetchone()
|
|
|
|
if result and result[0] == 2:
|
|
logger.info("query_history表更新成功")
|
|
else:
|
|
logger.error(f"query_history表更新失败，当前值: {result[0] if result else '记录不存在'}")
|
|
return False
|
|
|
|
# 检查car_owners表是否保存成功
|
|
for car_id in car_owner_ids:
|
|
cursor.execute("SELECT plate_number, query_history_id FROM car_owners WHERE id = ?", (car_id,))
|
|
result = cursor.fetchone()
|
|
|
|
if result and result[1] == query_id:
|
|
logger.info(f"car_owners表记录 {car_id} (车牌号: {result[0]}) 保存成功，并正确关联了查询历史")
|
|
else:
|
|
logger.error(f"car_owners表记录 {car_id} 保存失败或关联错误")
|
|
return False
|
|
|
|
conn.close()
|
|
return True
|
|
|
|
except sqlite3.Error as e:
|
|
logger.error(f"验证数据时发生数据库错误: {str(e)}")
|
|
return False
|
|
except Exception as e:
|
|
logger.error(f"验证数据时发生错误: {str(e)}")
|
|
return False
|
|
|
|
def main():
|
|
"""
|
|
主函数
|
|
"""
|
|
logger.info("开始最终验证")
|
|
|
|
# 创建测试数据
|
|
query_id = create_test_data()
|
|
if not query_id:
|
|
logger.error("无法创建测试数据，验证失败")
|
|
return
|
|
|
|
# 测试update_query_result接口
|
|
success, result = test_update_query_result_with_real_data(query_id)
|
|
if not success:
|
|
logger.error("update_query_result接口测试失败")
|
|
return
|
|
|
|
# 验证数据是否正确保存
|
|
if result and 'data' in result and 'car_owner_ids' in result['data']:
|
|
car_owner_ids = result['data']['car_owner_ids']
|
|
verify_success = verify_data_in_db(query_id, car_owner_ids)
|
|
|
|
if verify_success:
|
|
logger.info("所有验证通过！系统功能正常工作")
|
|
else:
|
|
logger.error("数据验证失败")
|
|
else:
|
|
logger.error("响应数据中不包含car_owner_ids字段")
|
|
|
|
logger.info("最终验证完成")
|
|
|
|
if __name__ == "__main__":
|
|
main() |