#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 科技部驾驶舱接口测试脚本 测试所有13个接口，使用2025年12月的数据 """ import subprocess import json import sys from datetime import datetime # 配置 BASE_URL = "http://localhost:2011" API_BASE = f"{BASE_URL}/api/Extend/LqTechDepartmentDashboard" TECH_DEPT_ID = "734725579919590661" # 科技一部ID STATISTICS_MONTH = "202512" # 2025年12月 # 测试结果 test_results = [] def get_token(): """获取登录token""" cmd = [ 'curl', '-s', '-X', 'POST', f'{BASE_URL}/api/oauth/Login', '-H', 'Content-Type: application/x-www-form-urlencoded', '-d', 'account=admin&password=e10adc3949ba59abbe56e057f20f883e' ] try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=10) if result.returncode == 0: data = json.loads(result.stdout) if data.get('code') == 200: token = data.get('data', {}).get('token', '') return token return None except Exception as e: print(f"获取Token失败: {e}") return None def test_api(name, endpoint, method="POST", data=None): """测试API接口""" token = get_token() if not token: return {"name": name, "endpoint": endpoint, "success": False, "error": "Token获取失败"} headers = [ 'Authorization: ' + token, 'Content-Type: application/json' ] cmd = ['curl', '-s', '-X', method] for h in headers: cmd.extend(['-H', h]) if data: cmd.extend(['-d', json.dumps(data)]) cmd.append(f"{API_BASE}/{endpoint}") try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) if result.returncode == 0: try: response = json.loads(result.stdout) success = response.get('code') == 200 or 'list' in response or 'TrendData' in response or 'RankingData' in response or 'DistributionData' in response return { "name": name, "endpoint": endpoint, "success": success, "status_code": response.get('code', 'N/A'), "has_data": 'data' in response or 'list' in response, "error": None if success else response.get('msg', '未知错误') } except json.JSONDecodeError: return { "name": name, "endpoint": endpoint, "success": False, "error": f"JSON解析失败: {result.stdout[:100]}" } else: return { "name": name, "endpoint": endpoint, "success": False, "error": f"请求失败: {result.stderr}" } except subprocess.TimeoutExpired: return { "name": name, "endpoint": endpoint, "success": False, "error": "请求超时" } except Exception as e: return { "name": name, "endpoint": endpoint, "success": False, "error": str(e) } def main(): """主测试函数""" print("=" * 70) print("科技部驾驶舱接口测试") print("=" * 70) print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"测试月份: {STATISTICS_MONTH}") print(f"科技部ID: {TECH_DEPT_ID}") print("=" * 70) print() # 基础请求参数 base_input = { "techDepartmentId": TECH_DEPT_ID, "statisticsMonth": STATISTICS_MONTH } # 定义所有接口 apis = [ ("GetStatistics", "GetStatistics", "POST", base_input), ("GetShareStatistics", "GetShareStatistics", "POST", base_input), ("GetPerformanceTrend", "GetPerformanceTrend", "POST", {**base_input, "monthCount": 12}), ("GetShareTrend", "GetShareTrend", "POST", {**base_input, "monthCount": 12}), ("GetStoreRanking", "GetStoreRanking", "POST", base_input), ("GetStoreDistribution", "GetStoreDistribution", "POST", base_input), ("GetTeacherRanking", "GetTeacherRanking", "POST", base_input), ("GetOperationStatistics", "GetOperationStatistics", "POST", base_input), ("GetComparisonAnalysis", "GetComparisonAnalysis", "POST", base_input), ("GetStoreDetailList", "GetStoreDetailList", "POST", {**base_input, "currentPage": 1, "pageSize": 10}), ("GetTeacherDetailList", "GetTeacherDetailList", "POST", {**base_input, "currentPage": 1, "pageSize": 10}), ("GetBillingDetailList", "GetBillingDetailList", "POST", {**base_input, "currentPage": 1, "pageSize": 10}), ("GetConsumeDetailList", "GetConsumeDetailList", "POST", {**base_input, "currentPage": 1, "pageSize": 10}), ] # 测试每个接口 print("开始测试接口...") print() for name, endpoint, method, data in apis: print(f"测试 {name}...", end=" ", flush=True) result = test_api(name, endpoint, method, data) test_results.append(result) if result['success']: print("✅ 成功") else: print(f"❌ 失败: {result.get('error', '未知错误')}") # 输出测试报告 print() print("=" * 70) print("测试结果汇总") print("=" * 70) success_count = sum(1 for r in test_results if r['success']) total_count = len(test_results) print(f"总计: {total_count} 个接口") print(f"成功: {success_count} 个") print(f"失败: {total_count - success_count} 个") print() # 详细结果 print("详细结果:") for result in test_results: status = "✅" if result['success'] else "❌" print(f"{status} {result['name']}: {result['endpoint']}") if not result['success']: print(f" 错误: {result.get('error', '未知错误')}") print() print("=" * 70) # 生成测试报告文件 report_lines = [ "# 科技部驾驶舱接口测试报告", "", f"**测试时间**：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}", f"**测试月份**：{STATISTICS_MONTH}", "", "---", "", "## 接口测试结果", "" ] for result in test_results: report_lines.extend([ f"### {result['name']}", "", f"**接口**：`POST /api/Extend/LqTechDepartmentDashboard/{result['endpoint']}`", "", f"**接口说明**：{result['name']}", "", f"**接口请求参数**：", "```json", json.dumps(apis[[a[0] for a in apis].index(result['name'])][3], ensure_ascii=False, indent=2), "```", "", f"**接口返回结果**：{'一致' if result['success'] else '失败'}", "", f"**接口是否报错**：{'否' if result['success'] else '是'}", "", "" ]) report_content = "\n".join(report_lines) with open('docs/科技部驾驶舱接口测试报告.md', 'w', encoding='utf-8') as f: f.write(report_content) print(f"测试报告已保存到: docs/科技部驾驶舱接口测试报告.md") return 0 if success_count == total_count else 1 if __name__ == '__main__': sys.exit(main())