用python做自动化测试--对服务器端的自动化测试（2）-发送端模拟器

标签：python   自动化测试   http协议   

python的http client库很多，有httplib, urllib 和urllib2，但这几个用起来还是比较麻烦，requests 是高富帅（http://docs.python-requests.org/en/latest/），接口更简洁，优雅。 支持Json, 很方便设置发送的header， session管理。

#!/usr/bin/env python
#coding=utf-8

import time,logging,sys,os
import pickle
import uuid
import requests
import ConfigParser


class RequestClient:
    def __init__(self,client_cfg,mms_cfg_dir):
        #http://127.0.0.1:8080/mm7process
        self.client_cfg=client_cfg
        self.url=self.client_cfg.get("sysparm","peer_ip")+":"+                 self.client_cfg.get("sysparm","peer_port")+"/"
        self.mms_cfg_dir=mms_cfg_dir

    def get_path_data(self,fi,wifi_para):
        fi=open(fi,"r")
        lines=fi.readlines()
        fi.close()
        path=lines[0].strip().split(":")[1]
        data=lines[1].strip().split(":")[1]
        #这里是替换模板中的两个变量
        for k, v in wifi_para.iteritems():
            origi_data=origi_data.replace("{%"+k+"%}",v)
        logging.debug("path:"+path)
        logging.debug("post original data:"+origi_data)
        return path, data

    def wifi_client(self,send_msg_type,msg_id=None,wifi_kw={}):
        if send_msg_type=="reg":
            session = requests.session()
            conf_file=os.path.join(self.mms_cfg_dir,"registration.cfg")
            response=requests.post(self.url+"registration", data=self.get_path_data(conf_file,wifi_kw)[1])
            #print response.cookies
            #保存cookie为下面的消息流程使用，当然这里其实可以和下面的消息合并，提高程序的运行效率，但为了同时支持功能测试，方便功能测试的使用，
            #所以保存到文件了，对效率有一定牺牲
            cookie_file=open('cookie', 'wb')
            pickle.dump(response.cookies, cookie_file)
            cookie_file.close()

        elif send_msg_type=="smsv":
            conf_file=os.path.join(self.mms_cfg_dir,"smsverification.cfg")
            in_cookie_file=open('cookie', 'rb')
            in_cookies = pickle.load(in_cookie_file)
            in_cookie_file.close()
            response=requests.post(self.url+self.get_path_data(conf_file,wifi_kw)[0], data=self.get_path_data(conf_file,wifi_kw)[1],cookies=in_cookies)

        elif send_msg_type=="qba":
            conf_file=os.path.join(self.mms_cfg_dir,"querybalance.cfg")
            response=requests.post(self.url+self.get_path_data(conf_file,wifi_kw)[0], data=self.get_path_data(conf_file,wifi_kw)[1])

        else:
            logging.error("wrong command")

        logging.debug("receive response:"+response.text)


def init_attri_config_file(file_name):
    '''
    Use ConfigParser to parse below configuration file:
    [selection]:
    option:value
    '''
    config = ConfigParser.ConfigParser()
    try:
        if os.path.exists(file_name):
            config.read(file_name)
            return config
    except:
        logging.error(file_name+" is not exit")

def wifi_client_simulator(MM7Client):
    logging.debug("Start to do performace testing for wifi")
    MSISDN=2341824988270
    #complte message flow per second
    rate=10
    while 1:
        try:
            MSISDN=MSISDN+1
            logging.info("start testing with MSISDN "+str(MSISDN))
            wifi_parameter={}
            wifi_parameter["SessionID"]=str(uuid.uuid1())
            wifi_parameter["MSISDN"]=str(MSISDN)
            #这里打印每个消息发送和完成的时间，目的是为了在性能测试中，分析系统性能瓶颈，用这种方式，可以很快定位到是被测试系统瓶颈
            #还是模拟器瓶颈，或者是网络瓶颈。具体这块怎么分析，在后面专门做个性能测试分析的专题吧
            logging.info("Start register for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            MM7Client.mm7_client("reg",wifi_kw=wifi_parameter)
            logging.info("End register for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            smscode="1234"

            wifi_parameter["SMSCode"]=smscode
            logging.info("Start verification for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            MM7Client.mm7_client("smsv",wifi_kw=wifi_parameter)
            logging.info("End verification for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            #这里的用户名，密码，还有上面的smscode其实是从数据库获得的，为了代码演示，这里直接赋值了
            user_password=("test","testpasswd")
            logging.debug("username and password"+str(user_password))
            wifi_parameter["UserName"]=str(user_password[0])
            wifi_parameter["UserPassword"]=str(user_password[1])
            logging.info("Start qba for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            MM7Client.mm7_client("qba",wifi_kw=wifi_parameter)
            logging.info("End qba for  MSISDN "+str(MSISDN)+" at "+str(time.time()))
            logging.info("Ended testing with MSISDN "+str(MSISDN))
            time.sleep(1.0/rate)

        except:
            logging.exception("exception:")


if __name__=="__main__":
    import sys
    sys.setdefaultencoding('utf-8')
    logging.basicConfig(filename="client.log",level=0,format='%(asctime)s %(name)s %(levelname)s %(module)s:%(lineno)d %(message)s')
    mms_cfg_dir="D:\jobs\code\TestScript\SVN\wificlient\conf"
    cfg=init_attri_config_file(os.path.join(mms_cfg_dir,"mmsc.cfg"))
    Wifi_request_client=RequestClient(cfg,mms_cfg_dir)
    wifi_client_simulator(Wifi_request_client)

配置文件例子，mmsc.cfg：

[sysparm]
#以下2个是gateway接受消息的地址。
peer_ip: http://192.168.53.20
peer_port:  8991

registration.cfg文件例子：

path:webauthv2/api/v1/user/register?ssid=6721&phase=1
data:SessionID={%SessionID%}&MessageType=1&MSISDN={%MSISDN%}&Platform=iOS&MACAddress=001122334455

在这里，我们需要保证每次生成的SessionID 和MSISDN不一样，所以每次需要利用此模板，替换这个2个值。

这样一个http client的模拟器就完成了，持续的给服务器发送消息。如果要高并发的，可以改成多线程或者多进程方式。

用python做自动化测试--对服务器端的自动化测试（2）-发送端模拟器,布布扣,bubuko.com

用python做自动化测试--对服务器端的自动化测试（2）-发送端模拟器

标签：python   自动化测试   http协议   

原文地址：http://blog.csdn.net/powerccna/article/details/38083479