#!/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())