1. 前言
由于公司的一个项目是基于B/S架构与WEB服务通信，使用XML数据作为通信数据，在添加新功能时，WEB端与客户端分别由不同的部门负责，所以在WEB端功能实现过程中，需要自己发起请求测试，于是便选择了使用Python编写此脚本。另外由于此脚本最开始希望能在以后发展成具有压力测试的功能，所以除了基本的访问之外，添加了多线程请求。
整个脚本主要涉及到的关于Python的知识点包括：
基于urllib.request的Http访问
多线程
类与方法的定义
全局变量的定义与使用
文件的读取与写入
……
2. 源码与结果
整个程序包括Python源码和配置文件，由于源码中有相应的注释，所以就直接贴源码吧，如下：
|
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
|
# TradeWeb测试脚本
import threading, time, http.client, urllib.request, os
#import matplotlib.pyplot as plt
TOTAL = 0; # 总数
SUCC = 0; # 响应成功数量
FAIL = 0; # 响应失败数量
EXCEPT = 0 # 响应异常数
MAXTIME = 0 # 最大响应时间
MINTIME = 100 # 最小响应时间，初始值为100秒
COUNT_TIME = 0 # 总时间
THREAD_COUNT = 0 # 记录线程数量
CODE_MAP = {200:0, 301:0, 302:0, 304:0} # 状态码信息(部分)
RESULT_FILE = 'tradeWebResult.xml' # 输出结果文件
REQUEST_DATA_FILE = 'requestData.config' # 数据文件
DATA = '''请在tradeWebRequestData.config文件中配置'''
TIME_LIST = [] # 记录访问时间
#创建一个threading.Thread的派生类
class RequestThread(threading.Thread):
#构造函数
def __init__(self, thread_name):
threading.Thread.__init__(self)
self.test_count = 0;
#线程运行的入口函数
def run(self):
global THREAD_COUNT
THREAD_COUNT += 1
#print("Start the count of thread:%d" %(THREAD_COUNT))
self.testPerformace()
#测试性能方法
def testPerformace(self):
global TOTAL
global SUCC
global FAIL
global EXCEPT
global DATA
global COUNT_TIME
global CODE_MAP
global URL
try:
st = time.time() #记录开始时间
start_time
cookies = urllib.request.HTTPCookieProcessor()
opener = urllib.request.build_opener(cookies)
resp = urllib.request.Request(url=URL,
headers={'Content-Type':'text/xml', 'Connection':'Keep-Alive'},
data=DATA.encode('gbk'))
respResult = opener.open(resp)
# 记录状态码 START
code = respResult.getcode()
if code == 200:
SUCC += 1
else:
FAIL += 1
if code in CODE_MAP.keys():
CODE_MAP[code] += 1
else:
CODE_MAP[code] = 1
# print(request.status)
# 记录状态码 END
html = respResult.read().decode('gbk')
print(html)
time_span = time.time() - st # 计算访问时间
# 记录访问时间
TIME_LIST.append(round(time_span * 1000))
# print('%-13s: %f ' %(self.name, time_span))
self.maxtime(time_span)
self.mintime(time_span)
self.writeToFile(html)
# info = respResult.info() # 响应头信息
# url = respResult.geturl() # URL地址
# print(info);
# print(url)
COUNT_TIME += time_span
TOTAL += 1
except Exception as e:
print(e)
TOTAL += 1
EXCEPT += 1
# 设置最大时间，如果传入的时间大于当前最大时间
def maxtime(self, ts):
global MAXTIME
#print("time:%f" %(ts))
if ts > MAXTIME:
MAXTIME = ts
# 设置最小时间，如果传入的时间小于当前最小时间
def mintime(self, ts):
global MINTIME
#print("time:%f" %(ts))
if ts < MINTIME and ts > 0.000000000000000001:
MINTIME = ts
# 写入文件
def writeToFile(self, html):
f = open(RESULT_FILE, 'w')
f.write(html)
f.write('\r\n')
f.close();
# 读取XML数据信息
def loadData():
global URL
global DATA
f = open(REQUEST_DATA_FILE, 'r')
URL = "".join(f.readline())
DATA = "".join(f.readlines())
# print(DATA)
f.close()
if __name__ == "__main__":
# print("============测试开始============")
print("")
# 开始时间
start_time = time.time()
# 并发的线程数
thread_count = 1
loadData() # 加载请求数据
i = 0
while i < thread_count:
t = RequestThread("Thread" + str(i))
t.start()
i += 1
t = 0
while TOTAL < thread_count and t < 60:
# print("total:%d, succ:%d, fail:%d, except:%d\n" %(TOTAL,SUCC,FAIL,EXCEPT))
print("正在请求 ",URL)
t += 1
time.sleep(1)
# 打印信息
print()
print("请求", URL, "的统计信息：")
print(" 总请求数 = %d，成功 = %d，失败 = %d，异常 = %d" %(TOTAL, SUCC, FAIL, EXCEPT))
print()
print("往返程的估计时间(以毫秒为单位)：")
print(" 合计 =", int(COUNT_TIME * 1000), "ms", end = '')
print(" 最大 =", round(MAXTIME * 1000), "ms", end = '')
print(" 最小 =", round(MINTIME * 1000), "ms", end = '')
print(" 平均 =", round((COUNT_TIME / thread_count) * 1000), "ms")
print()
print("响应的状态码与次数信息(状态码:次数)：")
print(" ", CODE_MAP)
print()
print("输出页面请查看", RESULT_FILE, "文件(建议使用浏览器或XML专业工具打开)")
print()
# os.system("pause")
print(TIME_LIST)
input()
|
配置文件主要在于易于更改访问路径等，其中SESSION_ID是在Fiddler中抓包获取，配置文件源文件如下（为不泄露公司隐私，数据并非原始数据，但格式相同）：
|
1
2
3
4
5
6
7
8
9
10
|
http://127.0.0.1:8888/XXXXXXXXX/httpXmlServlet
<?xml version=“1.0” encoding = “GB2312”?>
<COM>
<REQ name="commodity_query">
<USER_ID>0001</USER_ID>
<COMMODITY_ID>0000</COMMODITY_ID>
<SESSION_ID>4918081208706966071</SESSION_ID>
</REQ>
</COM>
|
测试结果如下：
由于公司保密性要求，地址做了模糊处理，另外输出的tradeWebResult.xml结果页面也未展示。
以上仅为个人学习与使用Python过程的一个记录，难免会有程序设计或使用不当，如有更好的意见，欢迎指正。
注：此代码开发环境为Python 3.5 + windows，未在Python 2.x环境下测试
以上这篇Python 3.x基于Xml数据的Http请求方法就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持服务器之家。
原文链接：https://blog.csdn.net/t894690230/article/details/50598718