110 lines
3.9 KiB
Python
110 lines
3.9 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""验证所有7个接口是否正常工作"""
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
# 添加项目根目录到 Python 路径
|
||
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
||
|
|
|
||
|
|
def test_interfaces():
|
||
|
|
"""测试所有接口"""
|
||
|
|
from services.token_service import TokenService
|
||
|
|
from services.file_service import FileService
|
||
|
|
from services.statement_service import StatementService
|
||
|
|
from utils.error_simulator import ErrorSimulator
|
||
|
|
|
||
|
|
print("=" * 60)
|
||
|
|
print("Interface Alignment Verification Test")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
# 1. 验证 TokenService
|
||
|
|
print("\n[1/6] TokenService initialization...")
|
||
|
|
token_svc = TokenService()
|
||
|
|
print(" [OK] TokenService initialized")
|
||
|
|
|
||
|
|
# 2. 验证 FileService
|
||
|
|
print("\n[2/6] FileService initialization...")
|
||
|
|
file_svc = FileService()
|
||
|
|
print(" [OK] FileService initialized")
|
||
|
|
|
||
|
|
# 3. 验证 StatementService
|
||
|
|
print("\n[3/6] StatementService initialization...")
|
||
|
|
stmt_svc = StatementService()
|
||
|
|
print(" [OK] StatementService initialized")
|
||
|
|
|
||
|
|
# 4. 验证错误码
|
||
|
|
print("\n[4/6] Error codes verification...")
|
||
|
|
assert "40100" in ErrorSimulator.ERROR_CODES, "Error code 40100 not found"
|
||
|
|
assert ErrorSimulator.ERROR_CODES["40100"]["message"] == "未知异常", "Error message incorrect"
|
||
|
|
print(" [OK] Error code 40100 added")
|
||
|
|
|
||
|
|
# 5. 验证响应模板文件
|
||
|
|
print("\n[5/6] Response template files verification...")
|
||
|
|
import json
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
responses_dir = Path("config/responses")
|
||
|
|
|
||
|
|
# 检查 token.json
|
||
|
|
with open(responses_dir / "token.json", encoding='utf-8') as f:
|
||
|
|
token_data = json.load(f)
|
||
|
|
assert isinstance(token_data["success_response"]["data"]["analysisType"], int), "analysisType should be integer"
|
||
|
|
print(" [OK] token.json format correct (analysisType is integer)")
|
||
|
|
|
||
|
|
# 检查 upload_status.json
|
||
|
|
assert (responses_dir / "upload_status.json").exists(), "upload_status.json not found"
|
||
|
|
print(" [OK] upload_status.json created")
|
||
|
|
|
||
|
|
# 检查 bank_statement.json
|
||
|
|
with open(responses_dir / "bank_statement.json", encoding='utf-8') as f:
|
||
|
|
stmt_data = json.load(f)
|
||
|
|
assert len(stmt_data["success_response"]["data"]["bankStatementList"]) > 0, "bankStatementList is empty"
|
||
|
|
print(" [OK] bank_statement.json format correct")
|
||
|
|
|
||
|
|
# 6. 验证 FileRecord 字段
|
||
|
|
print("\n[6/6] FileRecord fields verification...")
|
||
|
|
from services.file_service import FileRecord
|
||
|
|
|
||
|
|
record = FileRecord(
|
||
|
|
log_id=10001,
|
||
|
|
group_id=1000,
|
||
|
|
file_name="test.csv"
|
||
|
|
)
|
||
|
|
|
||
|
|
# 检查所有必需字段是否存在
|
||
|
|
required_fields = [
|
||
|
|
'account_no_list', 'enterprise_name_list', 'bank_name', 'real_bank_name',
|
||
|
|
'template_name', 'data_type_info', 'file_size', 'download_file_name',
|
||
|
|
'file_package_id', 'file_upload_by', 'file_upload_by_user_name',
|
||
|
|
'file_upload_time', 'le_id', 'login_le_id', 'log_type', 'log_meta',
|
||
|
|
'lost_header', 'rows', 'source', 'total_records', 'is_split',
|
||
|
|
'trx_date_start_id', 'trx_date_end_id'
|
||
|
|
]
|
||
|
|
|
||
|
|
for field in required_fields:
|
||
|
|
assert hasattr(record, field), f"FileRecord missing field: {field}"
|
||
|
|
|
||
|
|
print(" [OK] FileRecord contains all {} required fields".format(len(required_fields)))
|
||
|
|
|
||
|
|
print("\n" + "=" * 60)
|
||
|
|
print("[SUCCESS] All verifications passed!")
|
||
|
|
print("=" * 60)
|
||
|
|
|
||
|
|
print("\nInterface List:")
|
||
|
|
print("1. POST /account/common/getToken")
|
||
|
|
print("2. POST /watson/api/project/remoteUploadSplitFile")
|
||
|
|
print("3. POST /watson/api/project/getJZFileOrZjrcuFile")
|
||
|
|
print("4. POST /watson/api/project/upload/getpendings")
|
||
|
|
print("5. GET /watson/api/project/bs/upload [NEW]")
|
||
|
|
print("6. POST /watson/api/project/batchDeleteUploadFile")
|
||
|
|
print("7. POST /watson/api/project/getBSByLogId")
|
||
|
|
|
||
|
|
print("\nNext Steps:")
|
||
|
|
print("- Run: python main.py")
|
||
|
|
print("- Visit: http://localhost:8000/docs")
|
||
|
|
print("- Test all 7 interfaces")
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
test_interfaces()
|