init

tests/__init__.py

@@ -0,0 +1 @@
+# Tests package

tests/conftest.py

@@ -0,0 +1,48 @@
+"""
+Pytest 配置和共享 fixtures
+"""
+import pytest
+from fastapi.testclient import TestClient
+import sys
+import os
+
+# 添加项目根目录到 sys.path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from main import app
+
+
+@pytest.fixture
+def client():
+    """创建测试客户端"""
+    return TestClient(app)
+
+
+@pytest.fixture
+def sample_token_request():
+    """示例 Token 请求 - 返回 form-data 格式的数据"""
+    return {
+        "projectNo": "test_project_001",
+        "entityName": "测试企业",
+        "userId": "902001",
+        "userName": "902001",
+        "appId": "remote_app",
+        "appSecretCode": "test_secret_code_12345",
+        "role": "VIEWER",
+        "orgCode": "902000",
+        "departmentCode": "902000",
+    }
+
+
+@pytest.fixture
+def sample_inner_flow_request():
+    """示例拉取行内流水请求"""
+    return {
+        "groupId": 1001,
+        "customerNo": "test_customer_001",
+        "dataChannelCode": "test_code",
+        "requestDateId": 20240101,
+        "dataStartDateId": 20240101,
+        "dataEndDateId": 20240131,
+        "uploadUserId": 902001,
+    }

tests/integration/__init__.py

@@ -0,0 +1 @@
+# Integration tests package

tests/integration/test_full_workflow.py

@@ -0,0 +1,125 @@
+"""
+集成测试 - 完整的接口调用流程测试
+"""
+import pytest
+import time
+
+
+def test_complete_workflow(client):
+    """测试完整的接口调用流程"""
+    # 1. 获取 Token
+    response = client.post(
+        "/account/common/getToken",
+        data={
+            "projectNo": "integration_test_001",
+            "entityName": "集成测试企业",
+            "userId": "902001",
+            "userName": "902001",
+            "appId": "remote_app",
+            "appSecretCode": "test_secret_code_12345",
+            "role": "VIEWER",
+            "orgCode": "902000",
+            "departmentCode": "902000",
+        },
+    )
+    assert response.status_code == 200
+    token_data = response.json()
+    assert token_data["code"] == "200"
+    project_id = token_data["data"]["projectId"]
+    token = token_data["data"]["token"]
+    assert token is not None
+
+    # 2. 上传文件（模拟）
+    # 注意：在测试环境中，我们跳过实际的文件上传，直接测试其他接口
+
+    # 3. 检查解析状态
+    response = client.post(
+        "/watson/api/project/upload/getpendings",
+        data={"groupId": project_id, "inprogressList": "10001"},
+    )
+    assert response.status_code == 200
+    status_data = response.json()
+    assert "parsing" in status_data["data"]
+
+    # 4. 获取银行流水
+    response = client.post(
+        "/watson/api/project/getBSByLogId",
+        data={
+            "groupId": project_id,
+            "logId": 10001,
+            "pageNow": 1,
+            "pageSize": 10,
+        },
+    )
+    assert response.status_code == 200
+    statement_data = response.json()
+    assert statement_data["code"] == "200"
+    assert "bankStatementList" in statement_data["data"]
+    assert "totalCount" in statement_data["data"]
+
+
+def test_all_error_codes(client):
+    """测试所有错误码"""
+    error_codes = ["40101", "40102", "40104", "40105", "40106", "40107", "40108"]
+
+    for error_code in error_codes:
+        response = client.post(
+            "/account/common/getToken",
+            data={
+                "projectNo": f"test_error_{error_code}",
+                "entityName": "测试企业",
+                "userId": "902001",
+                "userName": "902001",
+                "appId": "remote_app",
+                "appSecretCode": "test_secret_code_12345",
+                "role": "VIEWER",
+                "orgCode": "902000",
+                "departmentCode": "902000",
+            },
+        )
+        assert response.status_code == 200
+        data = response.json()
+        assert data["code"] == error_code, f"错误码 {error_code} 未正确触发"
+        assert data["successResponse"] == False
+
+
+def test_pagination(client):
+    """测试分页功能"""
+    # 获取 Token
+    response = client.post(
+        "/account/common/getToken",
+        data={
+            "projectNo": "pagination_test",
+            "entityName": "分页测试",
+            "userId": "902001",
+            "userName": "902001",
+            "appId": "remote_app",
+            "appSecretCode": "test_secret_code_12345",
+            "role": "VIEWER",
+            "orgCode": "902000",
+            "departmentCode": "902000",
+        },
+    )
+    project_id = response.json()["data"]["projectId"]
+
+    # 测试第一页
+    response = client.post(
+        "/watson/api/project/getBSByLogId",
+        data={"groupId": project_id, "logId": 10001, "pageNow": 1, "pageSize": 1},
+    )
+    page1 = response.json()
+
+    # 测试第二页
+    response = client.post(
+        "/watson/api/project/getBSByLogId",
+        data={"groupId": project_id, "logId": 10001, "pageNow": 2, "pageSize": 1},
+    )
+    page2 = response.json()
+
+    # 验证总记录数相同
+    assert page1["data"]["totalCount"] == page2["data"]["totalCount"]
+
+    # 验证页码不同
+    if page1["data"]["totalCount"] > 1:
+        assert len(page1["data"]["bankStatementList"]) == 1
+        assert len(page2["data"]["bankStatementList"]) >= 0

tests/test_api.py

@@ -0,0 +1,176 @@
+"""
+API 端点测试
+"""
+
+
+def test_root_endpoint(client):
+    """测试根路径"""
+    response = client.get("/")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "running"
+    assert "swagger_docs" in data
+
+
+def test_health_check(client):
+    """测试健康检查端点"""
+    response = client.get("/health")
+    assert response.status_code == 200
+    data = response.json()
+    assert data["status"] == "healthy"
+
+
+def test_get_token_success(client, sample_token_request):
+    """测试获取 Token - 成功场景"""
+    response = client.post("/account/common/getToken", data=sample_token_request)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == "200"
+    assert "token" in data["data"]
+    assert "projectId" in data["data"]
+
+
+def test_get_token_error_40101(client):
+    """测试获取 Token - 错误场景 40101"""
+    request_data = {
+        "projectNo": "test_error_40101",
+        "entityName": "测试企业",
+        "userId": "902001",
+        "userName": "902001",
+        "appId": "remote_app",
+        "appSecretCode": "test_secret_code_12345",
+        "role": "VIEWER",
+        "orgCode": "902000",
+        "departmentCode": "902000",
+    }
+    response = client.post("/account/common/getToken", data=request_data)
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == "40101"
+    assert data["successResponse"] == False
+
+
+def test_fetch_inner_flow_success(client, sample_inner_flow_request):
+    """测试拉取行内流水 - 成功场景"""
+    response = client.post(
+        "/watson/api/project/getJZFileOrZjrcuFile",
+        data=sample_inner_flow_request
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == "200"
+    assert data["successResponse"] == True
+    assert isinstance(data["data"], list)
+    assert len(data["data"]) == 1
+    assert isinstance(data["data"][0], int)
+    assert 10000 <= data["data"][0] <= 99999
+
+
+def test_fetch_inner_flow_error_501014(client):
+    """测试拉取行内流水 - 错误场景 501014"""
+    request_data = {
+        "groupId": 1001,
+        "customerNo": "test_error_501014",
+        "dataChannelCode": "test_code",
+        "requestDateId": 20240101,
+        "dataStartDateId": 20240101,
+        "dataEndDateId": 20240131,
+        "uploadUserId": 902001,
+    }
+    response = client.post(
+        "/watson/api/project/getJZFileOrZjrcuFile",
+        data=request_data
+    )
+    assert response.status_code == 200
+    data = response.json()
+    assert data["code"] == "501014"
+    assert data["successResponse"] == False
+
+
+def test_get_upload_status_with_log_id(client):
+    """测试带 logId 参数查询返回非空 logs"""
+    response = client.get("/watson/api/project/bs/upload?groupId=1000&logId=13994")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # 验证基本响应结构
+    assert data["code"] == "200"
+    assert data["status"] == "200"
+    assert data["successResponse"] is True
+
+    # 验证 logs 不为空
+    assert len(data["data"]["logs"]) == 1
+
+    # 验证返回的 logId 正确
+    log = data["data"]["logs"][0]
+    assert log["logId"] == 13994
+
+    # 验证固定成功状态
+    assert log["status"] == -5
+    assert log["uploadStatusDesc"] == "data.wait.confirm.newaccount"
+
+    # 验证 logMeta 格式正确
+    assert log["logMeta"] == "{\"lostHeader\":[],\"balanceAmount\":\"-1\"}"
+
+
+def test_get_upload_status_without_log_id(client):
+    """测试不带 logId 参数查询返回空 logs 数组"""
+    response = client.get("/watson/api/project/bs/upload?groupId=1000")
+
+    assert response.status_code == 200
+    data = response.json()
+
+    # 验证基本响应结构
+    assert data["code"] == "200"
+    assert data["status"] == "200"
+    assert data["successResponse"] is True
+
+    # 验证 logs 为空
+    assert len(data["data"]["logs"]) == 0
+
+    # 验证其他字段存在
+    assert data["data"]["status"] == ""
+    assert data["data"]["accountId"] == 8954
+    assert data["data"]["currency"] == "CNY"
+
+
+def test_deterministic_data_generation(client):
+    """测试相同 logId 多次查询返回相同的核心字段值"""
+    # 第一次查询
+    response1 = client.get("/watson/api/project/bs/upload?groupId=1000&logId=13994")
+    log1 = response1.json()["data"]["logs"][0]
+
+    # 第二次查询
+    response2 = client.get("/watson/api/project/bs/upload?groupId=1000&logId=13994")
+    log2 = response2.json()["data"]["logs"][0]
+
+    # 验证关键字段相同
+    assert log1["logId"] == log2["logId"]
+    assert log1["bankName"] == log2["bankName"]
+    assert log1["accountNoList"] == log2["accountNoList"]
+    assert log1["enterpriseNameList"] == log2["enterpriseNameList"]
+    assert log1["status"] == log2["status"]
+    assert log1["logMeta"] == log2["logMeta"]
+    assert log1["templateName"] == log2["templateName"]
+    assert log1["trxDateStartId"] == log2["trxDateStartId"]
+    assert log1["trxDateEndId"] == log2["trxDateEndId"]
+
+
+def test_field_completeness(client):
+    """测试返回数据包含完整的 26 个字段"""
+    response = client.get("/watson/api/project/bs/upload?groupId=1000&logId=13994")
+    log = response.json()["data"]["logs"][0]
+
+    # 验证所有必需字段存在
+    required_fields = [
+        "accountNoList", "bankName", "dataTypeInfo", "downloadFileName",
+        "enterpriseNameList", "fileSize", "fileUploadBy", "fileUploadByUserName",
+        "fileUploadTime", "isSplit", "leId", "logId", "logMeta", "logType",
+        "loginLeId", "lostHeader", "realBankName", "rows", "source", "status",
+        "templateName", "totalRecords", "trxDateEndId", "trxDateStartId",
+        "uploadFileName", "uploadStatusDesc"
+    ]
+
+    for field in required_fields:
+        assert field in log, f"缺少字段: {field}"