test_tech_dashboard_apis.py
7.36 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
#!/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())