test_reimbursement_workflow_config_api.py 16.9 KB

Edit Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
报销流程配置接口测试脚本
用于测试报销流程配置相关的所有接口
"""

import requests
import json
import sys

# 配置
BASE_URL = "http://localhost:2011"
LOGIN_URL = f"{BASE_URL}/api/oauth/Login"
WORKFLOW_CONFIG_URL = f"{BASE_URL}/api/Extend/LqReimbursementWorkflowConfig"

# 测试结果
test_results = []

def log_test(name, success, message=""):
    """记录测试结果"""
    status = "✅ PASS" if success else "❌ FAIL"
    print(f"{status} - {name}")
    if message:
        print(f"    {message}")
    test_results.append({"name": name, "success": success, "message": message})

def get_token():
    """获取认证Token"""
    try:
        response = requests.post(
            LOGIN_URL,
            data={
                "account": "admin",
                "password": "e10adc3949ba59abbe56e057f20f883e"
            },
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        if result.get("code") == 200 and result.get("data") and result.get("data").get("token"):
            token = result["data"]["token"]
            print(f"✅ 获取Token成功")
            return token
        else:
            print(f"❌ 获取Token失败: {result}")
            return None
    except requests.exceptions.ConnectionError:
        print(f"❌ 无法连接到服务器 {BASE_URL}")
        print("   请确保后端服务已启动，并且运行在 http://localhost:2011")
        return None
    except Exception as e:
        print(f"❌ 获取Token异常: {str(e)}")
        return None

def test_get_enabled_list(token):
    """测试1: 获取启用的流程列表"""
    print("\n=== 测试1: 获取启用的流程列表 ===")
    try:
        response = requests.get(
            f"{WORKFLOW_CONFIG_URL}/Actions/GetEnabledList",
            headers={"Authorization": token},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200:
            data = result.get("data", {})
            list_data = data.get("list", [])
            log_test("获取启用的流程列表", True, f"返回 {len(list_data)} 个流程")
            return True
        else:
            log_test("获取启用的流程列表", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False
    except Exception as e:
        log_test("获取启用的流程列表", False, f"异常: {str(e)}")
        return False

def test_get_list(token):
    """测试2: 获取流程列表（分页）"""
    print("\n=== 测试2: 获取流程列表（分页） ===")
    try:
        response = requests.get(
            f"{WORKFLOW_CONFIG_URL}",
            params={
                "currentPage": 1,
                "pageSize": 20,
                "keyword": "",
                "queryJson": json.dumps({"isEnabled": 1}) if True else None
            },
            headers={"Authorization": token},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200:
            data = result.get("data", {})
            pagination = data.get("pagination", {})
            list_data = data.get("list", [])
            total = pagination.get("total", 0)
            log_test("获取流程列表", True, f"总数: {total}, 当前页: {len(list_data)} 条")
            return True, list_data
        else:
            log_test("获取流程列表", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False, None
    except Exception as e:
        log_test("获取流程列表", False, f"异常: {str(e)}")
        return False, None

def test_create_workflow(token):
    """测试3: 创建流程配置"""
    print("\n=== 测试3: 创建流程配置 ===")
    try:
        create_data = {
            "workflowName": "测试流程-" + str(int(__import__("time").time())),
            "isEnabled": 1,
            "description": "这是一个测试流程配置",
            "nodes": [
                {
                    "nodeOrder": 1,
                    "nodeName": "部门经理审批",
                    "approvalType": "会签",
                    "isRequired": 1,
                    "approverIds": [],
                    "approverNames": []
                },
                {
                    "nodeOrder": 2,
                    "nodeName": "财务审批",
                    "approvalType": "会签",
                    "isRequired": 1,
                    "approverIds": [],
                    "approverNames": []
                }
            ]
        }

        response = requests.post(
            f"{WORKFLOW_CONFIG_URL}",
            json=create_data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200:
            workflow_id = result.get("data", {}).get("id")
            if workflow_id:
                log_test("创建流程配置", True, f"创建成功, ID: {workflow_id}")
                return True, workflow_id, create_data.get("workflowName")
            else:
                log_test("创建流程配置", False, f"返回数据中没有ID: {result}")
                return False, None, None
        else:
            log_test("创建流程配置", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False, None, None
    except Exception as e:
        log_test("创建流程配置", False, f"异常: {str(e)}")
        return False, None, None

def test_get_info(token, workflow_id):
    """测试4: 获取流程详细信息"""
    print("\n=== 测试4: 获取流程详细信息 ===")
    if not workflow_id:
        log_test("获取流程详细信息", False, "没有可用的流程ID")
        return False

    try:
        response = requests.get(
            f"{WORKFLOW_CONFIG_URL}/{workflow_id}",
            headers={"Authorization": token},
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200:
            data = result.get("data", {})
            nodes = data.get("nodes", [])
            log_test("获取流程详细信息", True, f"流程名称: {data.get('workflowName')}, 节点数: {len(nodes)}")

            # 验证节点信息
            if nodes:
                first_node = nodes[0]
                log_test("节点信息完整性", True, f"第一个节点: {first_node.get('nodeName')}, 顺序: {first_node.get('nodeOrder')}")
            else:
                log_test("节点信息完整性", False, "没有节点信息")

            return True
        else:
            log_test("获取流程详细信息", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False
    except Exception as e:
        log_test("获取流程详细信息", False, f"异常: {str(e)}")
        return False

def test_update_workflow(token, workflow_id, original_name):
    """测试5: 更新流程配置"""
    print("\n=== 测试5: 更新流程配置 ===")
    if not workflow_id:
        log_test("更新流程配置", False, "没有可用的流程ID")
        return False

    try:
        update_data = {
            "id": workflow_id,
            "workflowName": f"{original_name}-已修改",
            "isEnabled": 1,
            "description": "这是修改后的流程配置描述",
            "nodes": [
                {
                    "nodeOrder": 1,
                    "nodeName": "部门经理审批（已修改）",
                    "approvalType": "会签",
                    "isRequired": 1,
                    "approverIds": [],
                    "approverNames": []
                },
                {
                    "nodeOrder": 2,
                    "nodeName": "财务审批（已修改）",
                    "approvalType": "或签",
                    "isRequired": 1,
                    "approverIds": [],
                    "approverNames": []
                },
                {
                    "nodeOrder": 3,
                    "nodeName": "总经理审批（新增）",
                    "approvalType": "会签",
                    "isRequired": 1,
                    "approverIds": [],
                    "approverNames": []
                }
            ]
        }

        response = requests.put(
            f"{WORKFLOW_CONFIG_URL}/{workflow_id}",
            json=update_data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200 or response.status_code == 200:
            log_test("更新流程配置", True, f"更新成功")

            # 验证更新结果
            verify_response = requests.get(
                f"{WORKFLOW_CONFIG_URL}/{workflow_id}",
                headers={"Authorization": token},
                timeout=10
            )
            if verify_response.status_code == 200:
                verify_result = verify_response.json()
                if verify_result.get("code") == 200:
                    verify_data = verify_result.get("data", {})
                    nodes_count = len(verify_data.get("nodes", []))
                    if nodes_count == 3 and verify_data.get("workflowName", "").endswith("-已修改"):
                        log_test("验证更新结果", True, f"节点数已更新为: {nodes_count}, 名称已修改")
                        return True
                    else:
                        log_test("验证更新结果", False, f"节点数: {nodes_count}, 名称: {verify_data.get('workflowName')}")
                        return False

            return True
        else:
            log_test("更新流程配置", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False
    except Exception as e:
        log_test("更新流程配置", False, f"异常: {str(e)}")
        return False

def test_create_with_approvers(token):
    """测试6: 创建带审批人的流程配置"""
    print("\n=== 测试6: 创建带审批人的流程配置 ===")
    try:
        # 先获取用户列表（这里需要根据实际情况调整）
        # 暂时使用空的审批人列表，实际使用时需要真实的用户ID

        create_data = {
            "workflowName": "测试流程-带审批人-" + str(int(__import__("time").time())),
            "isEnabled": 1,
            "description": "这是一个带审批人的测试流程配置",
            "nodes": [
                {
                    "nodeOrder": 1,
                    "nodeName": "部门经理审批",
                    "approvalType": "会签",
                    "isRequired": 1,
                    "approverIds": [],  # 实际测试时需要填入真实的用户ID
                    "approverNames": []
                }
            ]
        }

        response = requests.post(
            f"{WORKFLOW_CONFIG_URL}",
            json=create_data,
            headers={
                "Authorization": token,
                "Content-Type": "application/json"
            },
            timeout=10
        )
        response.raise_for_status()
        result = response.json()

        if result.get("code") == 200:
            workflow_id = result.get("data", {}).get("id")
            log_test("创建带审批人的流程配置", True, f"创建成功, ID: {workflow_id}")
            return True, workflow_id
        else:
            log_test("创建带审批人的流程配置", False, f"返回code: {result.get('code')}, msg: {result.get('msg')}")
            return False, None
    except Exception as e:
        log_test("创建带审批人的流程配置", False, f"异常: {str(e)}")
        return False, None

def test_create_validation(token):
    """测试7: 测试创建时的参数验证"""
    print("\n=== 测试7: 测试创建时的参数验证 ===")

    # 测试7.1: 流程名称为空
    print("\n--- 测试7.1: 流程名称为空 ---")
    try:
        create_data = {
            "workflowName": "",
            "isEnabled": 1,
            "nodes": [{"nodeOrder": 1, "nodeName": "节点1", "approvalType": "会签"}]
        }
        response = requests.post(
            f"{WORKFLOW_CONFIG_URL}",
            json=create_data,
            headers={"Authorization": token, "Content-Type": "application/json"},
            timeout=10
        )
        result = response.json()
        if result.get("code") != 200 or "不能为空" in str(result.get("msg", "")):
            log_test("流程名称为空验证", True, f"正确拒绝: {result.get('msg')}")
        else:
            log_test("流程名称为空验证", False, f"未正确验证: {result}")
    except Exception as e:
        log_test("流程名称为空验证", False, f"异常: {str(e)}")

    # 测试7.2: 节点列表为空
    print("\n--- 测试7.2: 节点列表为空 ---")
    try:
        create_data = {
            "workflowName": "测试流程",
            "isEnabled": 1,
            "nodes": []
        }
        response = requests.post(
            f"{WORKFLOW_CONFIG_URL}",
            json=create_data,
            headers={"Authorization": token, "Content-Type": "application/json"},
            timeout=10
        )
        result = response.json()
        if result.get("code") != 200 or "至少需要" in str(result.get("msg", "")):
            log_test("节点列表为空验证", True, f"正确拒绝: {result.get('msg')}")
        else:
            log_test("节点列表为空验证", False, f"未正确验证: {result}")
    except Exception as e:
        log_test("节点列表为空验证", False, f"异常: {str(e)}")

    # 测试7.3: 节点顺序不连续
    print("\n--- 测试7.3: 节点顺序不连续 ---")
    try:
        create_data = {
            "workflowName": "测试流程",
            "isEnabled": 1,
            "nodes": [
                {"nodeOrder": 1, "nodeName": "节点1", "approvalType": "会签"},
                {"nodeOrder": 3, "nodeName": "节点3", "approvalType": "会签"}  # 缺少节点2
            ]
        }
        response = requests.post(
            f"{WORKFLOW_CONFIG_URL}",
            json=create_data,
            headers={"Authorization": token, "Content-Type": "application/json"},
            timeout=10
        )
        result = response.json()
        if result.get("code") != 200 or "必须连续" in str(result.get("msg", "")):
            log_test("节点顺序不连续验证", True, f"正确拒绝: {result.get('msg')}")
        else:
            log_test("节点顺序不连续验证", False, f"未正确验证: {result}")
    except Exception as e:
        log_test("节点顺序不连续验证", False, f"异常: {str(e)}")

def main():
    """主测试函数"""
    print("=" * 60)
    print("报销流程配置接口测试")
    print("=" * 60)

    # 1. 获取Token
    print("\n步骤1: 获取认证Token...")
    token = get_token()
    if not token:
        print("\n❌ 无法获取Token，测试终止")
        sys.exit(1)

    # 2. 测试获取启用的流程列表
    test_get_enabled_list(token)

    # 3. 测试获取流程列表
    success, list_data = test_get_list(token)

    # 4. 测试创建流程配置
    create_success, workflow_id, workflow_name = test_create_workflow(token)

    # 5. 如果创建成功，测试获取详细信息
    if create_success and workflow_id:
        test_get_info(token, workflow_id)

        # 6. 测试更新流程配置
        test_update_workflow(token, workflow_id, workflow_name)

    # 7. 测试创建带审批人的流程配置
    test_create_with_approvers(token)

    # 8. 测试参数验证
    test_create_validation(token)

    # 9. 再次获取列表，验证创建和更新的结果
    print("\n=== 测试8: 验证创建和更新后的列表 ===")
    test_get_list(token)

    # 总结
    print("\n" + "=" * 60)
    print("测试总结")
    print("=" * 60)
    total = len(test_results)
    passed = sum(1 for r in test_results if r["success"])
    failed = total - passed

    print(f"总测试数: {total}")
    print(f"通过: {passed}")
    print(f"失败: {failed}")

    if failed > 0:
        print("\n失败的测试:")
        for r in test_results:
            if not r["success"]:
                print(f"  - {r['name']}: {r['message']}")

    print("=" * 60)

    if failed == 0:
        print("✅ 所有测试通过！")
        return 0
    else:
        print(f"❌ 有 {failed} 个测试失败")
        return 1

if __name__ == "__main__":
    sys.exit(main())