批量Post文件脚本

Zss 发表于:

需要测试约50个接口一起Post数据看是否正常,

这过程中发现每次post数据比较大的时候会占用非常大的内存,之前也没有释放内存,导致系统死了 ,再加上50个线程每次post几十MB的文件,物理内存一下子就非常高了,又是持续的post加上不是释放所以死了

请求时一个线程建一个 req = requests.Session() 去请求,请求完成后req.close()

最后del删除读取的data 和req,下次发再重新建

#coding:utf-8

import requests,time,os
from threading import Thread

def post_file(file_path,url,index,file_number):

    def format(number):
        if len(number) < 4:
            return ('%dB' % int(number))
        elif len(number) < 7:
            return ('%dKB' % (int(number) / 1000))
        else:
            return ('%dMB' % (int(number) / 1000000))
    file_size = format(str(os.path.getsize(file_path)))

    if '/' in file_path:
        file_name = file_path.split('/')[-1]
    else:
        file_name = file_path.split('\\')[-1]

    with open(file_path, 'rb') as f:
        data = {file_name: f.read()}
        f.close()
    while 1:
        req = requests.Session()
        try:
            rsp = req.post(url=url,files=data,timeout=30).status_code
            req.close()
            if rsp == 200:
                print('线程(%d-%d).FileSize:%s  Post_file:%s  Post_url:%s  Post_result: PASS!'%(index,file_number,file_size,file_name,url))
                break
            else:
                print('线程(%d-%d).FileSize:%s  Post_file:%s  Post_url:%s  Post_result: FAIL!  Erorr:%d'% (index,file_number,file_size,file_name, url,rsp))
                for i in range(1,4):
                    print('线程(%d-%d).第%d次重试...'%(index,file_number,i))
                    try:
                        rsp = req.post(url=url, files=data,timeout=30).status_code
                        req.close()
                        if rsp == 200:
                            print('线程(%d-%d).FileSize:%s  Post_file:%s  Post_url:%s  Post_result: PASS!' % (
                            index, file_number, file_size, file_name, url))
                            break
                        time.sleep(3)
                    except:
                        pass
                break

        except Exception as e:
            print('线程(%d-%d).FileSize:%s  Post_file:%s  Post_url:%s  Post_result: FAIL!  Erorr:%s' % (
            index, file_number, file_size, file_name, url,e))
            for i in range(1,4):
                try:
                    print('第%d次重试...' % i)
                    rsp = req.post(url=url, files=data,timeout=30).status_code
                    req.close()
                    if rsp == 200:
                        print('线程(%d-%d).FileSize:%s  Post_file:%s  Post_url:%s  Post_result: PASS!' % (index, file_number, file_size, file_name, url))
                        break
                    time.sleep(3)
                except:
                    pass
            break
    del data,req

def task(info,index):
    http_type,ip,port,username,password,file_list_path,number= info
    file_list = []
    for i, j, k in os.walk(file_list_path):
        for t in k:
            file_list.append(i + '\\' + t)
    url = '%s://%s:%s@%s:%s/upload' % (http_type, username, password, ip, port)
    print('线程(%d).检索 %s    文件数: %d个' % (index,file_list_path, len(file_list)))
    print('线程(%d).准备上传文件...'%(index))
    time.sleep(3)
    if number == '0':
        while 1:
            file_number = 1
            for file_path in file_list:
                post_file(file_path, url,index,file_number)
                file_number += 1
    else:
        while int(number) > 0:
            file_number = 1
            for file_path in file_list:
                post_file(file_path, url,index,file_number)
                file_number += 1
            number = int(number)-1

def run(info):
    T_list = []
    for i in range(len(info)):
        T = Thread(target=task,args=(info[i],i+1))
        T.start()
        T_list.append(T)
    for i in T_list:
        i.join()


if __name__ == '__main__':
    # (类型,ip,端口,账号,密码,需要上传的文件夹路径,传输次数)
    info = []
    for i in range(30001,30051):
        info.append(('http','192.168.4.10','%d'%i,'test','test',r'D:\file_list','0'))
    run(info)