任務目標：用requests構造一個上傳文件的接口測試，文件從一個文件夾里面取，可以并發上傳多個文件

一、分解任務：

①使用requests構造一個上傳文件的接口測試。

②實現遍歷上傳測試文件，作為并發上傳的效率對比。

③實現多線程上傳測試文件。

④從文件夾獲取該文件夾中全部文件的文件名，并以列表的方式存儲。

⑤對大數量的文件，實現指定線程數量的多線程上傳。

⑥生成測試報告。

二、使用requests構造一個上傳文件的接口測試。

①在cmd中輸入pip install requests，安裝測試所需的依賴

②調用requests中post方法，獲取接口返回數據。

三、實現遍歷上傳測試文件，作為并發上傳的效率對比。

創建testList，對testList中元素進行遍歷。

url = 'http://xxx'

params = {'fileName':'fileName','fileExt':'fileExt'}

testList1=['fileName1','fileName2','fileName3','fileName4','fileName5']??

#遍歷上傳：

print(ctime())

for i in range(len(testList1)):

??? files = {'file':open(testList1[i],'rb')}

??? r = requests.post(url,params,files=files)

??? print(i+1)

??? print("")

??? print(r.text)

print(ctime())

?四、實現多線程上傳測試文件。

①注意開頭要聲明 ?# -*- coding: UTF-8 -*- ? ，否則可能出現亂碼

②創建線程容器

③遍歷上傳上傳任務到線程容器

④啟動進程

#封裝send方法

def send(file):?

??? files = {'file':open(file,'rb')}???

??? r = requests.post(url,params,files=files)

??? print(r.text)

#多線程：

print(ctime())

threads = []

for file in testList2:

??? print(file+':')

??? t = threading.Thread(send(file))

??? threads.append(t)

?

if __name__ == '__main__':

??? #啟動進程

??? for i in range(len(testList)):

??????? threads[i].start()

??? for i in range(len(testList)):

??????? threads[i].join()?

print(ctime())

此時，已經能夠對遍歷和多線程方式的效率進行對比了。但由于數量太少，對比結果并不明顯，從我的結果來看，大致速度提升了2S。?我們需要大量的測試文件來對比。

五、從文件夾獲取該文件夾中全部文件的文件名，并以列表的方式存儲

①獲取文件名，以列表存儲

#獲取文件名
def file_name(path,testList): ??
? ? for root, dirs, files in os.walk(path): ?
? ? ? ? for file in files: ?
? ? ? ? ? ? if os.path.splitext(file)[1] == '.xlsx': ?
? ? ? ? ? ? ? ? testList.append(os.path.join(root, file))

?②將大列表拆分成小列表

thread_num = [i for i in range(len(testList2))]
n = 5 ? ? ?#定義小列表長度
list_temp = [testList2[i:i+n] for i in range(0,len(testList2),n)]

?六、對大數量的文件，實現指定線程數量的多線程上傳

上面定義了小列表長度為5。這里使用了兩個for循環，內層將小列表中的5個元素分配任務并裝入線程容器，外層則對小列表進行遍歷。（這里一定要注意啟動進程后，一定要創建新的線程容器，否則會報錯：RuntimeError: threads can only be started once）

threads = []
for temp in list_temp:
? ? #填充線程容器
? ? for file in temp:
? ? ? ? print(file+':')
? ? ? ? t = threading.Thread(target = send ,args = (file,))
? ? ? ? threads.append(t)

? ? if __name__ == '__main__':
? ? ? ? #啟動進程
? ? ? ? for i in range(len(temp)):
? ? ? ? ? ? threads[i].start()
? ? ? ? for i in range(len(temp)):
? ? ? ? ? ? threads[i].join()
? ? print('-------------------')
? ? print(ctime())

? ? #新線程容器
? ? threads = [] ????? ??? ? ? ? ? ? ? ? ? ? ? ? ?
print(ctime())

通過對500個測試文檔的轉換效率的測試，得出多線程的速度大致是單線程的n（n為線程數）倍?，理論測試數據量越大，越接近n值。n值的最優取值取決于計算機CPU單核的性能。

但針對于計算密集的CPU運算，則不適合采用多線程。假設單線程需要120S完成密集的運算工作，則多線程可能需要130S才能完成，原因是因為CPU的線程機制，具體什么機制百度有很詳細的解釋。而需要等待網絡響應的操作則特別適合多線程，因為計算機的響應速度，導致單線程會浪費很多的CPU資源，這時候使用多線程則能非常合理的利用到浪費的CPU資源，從而極大的提升運行效率。

?七、整合。

# -*- coding: UTF-8 -*-
import requests

from time import ctime,sleep
import threading
import os

url = 'http://xxx'
params = {'fileName':'xxx','fileExt':'xxx'}

#文件路徑
fileHomes1 = r"E:\\test\\excel1"
fileHomes2 = r"E:\\test\\excel2"

#獲取文件名
def file_name(path,testList): ??
? ? for root, dirs, files in os.walk(path): ?
? ? ? ? for file in files: ?
? ? ? ? ? ? if os.path.splitext(file)[1] == '.xlsx': ?
? ? ? ? ? ? ? ? testList.append(os.path.join(root, file))

#上傳：
def send(file):
? ? ? ? print("")
? ? ? ? files = {'file':open(file,'rb')} ? ?
? ? ? ? r = requests.post(url,params,files=files)
? ? ? ? print(r.text)

#定義testList
testList1=[] ??
testList2=[]
#賦值
file_name(fileHomes1,testList1)
file_name(fileHomes2,testList2)
#拆分列表
thread_num = [i for i in range(len(testList2))]
n = 5 ? ? ?#定義小列表長度
list_temp = [testList2[i:i+n] for i in range(0,len(testList2),n)]

print("遍歷上傳：")
print("----------------------------") ? ?
print(ctime())
print("----------------------------") ? ?
for i in range(len(testList1)): ? ? ?
? ? print(i+1)
? ? print("")
? ? send(testList1[i])
? ? print("") ? ?
print("----------------------------") ? ?
print(ctime())
print("----------------------------") ? ?

print('\n\n\n')
#多線程上傳
print("----------------------------") ? ?
print("多線程上傳:")
print("----------------------------")
print(ctime())
print("----------------------------")
#線程容器
threads = []
for temp in list_temp:
? ? #填充線程容器
? ? for file in temp:
? ? ? ? print(file+':')
? ? ? ? t = threading.Thread(target = send ,args = (file,))
? ? ? ? threads.append(t)

? ? if __name__ == '__main__':
? ? ? ? #啟動進程
? ? ? ? for i in range(len(temp)):
? ? ? ? ? ? threads[i].start()
? ? ? ? for i in range(len(temp)):
? ? ? ? ? ? threads[i].join()
? ? print('-------------------')
? ? print(ctime())
? ? #新容器
? ? threads = [] ??
? ? print('-------------------')
? ??? ? ? ? ? ? ? ? ? ? ? ? ?
print(ctime())
print("----------------------------")?

八、測試報告的生成

?

總結

以上是生活随笔為你收集整理的接口测试——并行上传文件的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。