小爸爸 发表于 2017-6-28 11:27:28

PICT 组合 +PageObject+Python asyncio 接口测试

说明
全对偶接口生成参数
功能
python3.4 asyncio aiohttp
unittest参数化
pageobject
数据维护用的YMAL
基于PICT全对偶生成接口参
代码分析
aiohttp asyncio 封装

<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>class request():
    def __init__(self, **kwargs):
      '''
      http请求的封装，传入dict
      :param req:
      '''
      self.req = kwargs
    def get(self, url, param):
      data = {}
      _url = self.req["protocol"] + self.req["host"] + ":" + str(self.req["port"]) + url
      print(_url +" get请求参数为:"+str(param))
      try:
            response = yield from aiohttp.request("GET", _url, headers=self.req["header"], params=param)
            string = (yield from response.read()).decode('utf-8')
            if response.status == 200:
                data = json.loads(string)
            else:
                print("data fetch failed for")
                print(response.content, response.status)
            data["status_code"] = response.status
            print(data)
      except asyncio.TimeoutError:
            print("访问失败")
      except UnicodeDecodeError:
            print("接口崩溃了")
      return data
    def post(self,url, param):
      data = {}
      _url = self.req["protocol"] + self.req["host"] + ':' + str(self.req["port"]) + url
      print(_url + " post接口参数为:" + str(param))
      try:
            response = yield from aiohttp.request('POST', _url, data=json.dumps(param), headers=self.req["header"])
            string = (yield from response.read()).decode('utf-8')
            if response.status == 200:
                data = json.loads(string)
            else:
                print("data fetch failed for")
                print(response.content, response.status)
            data["status_code"] = response.status
            print(data)
      except asyncio.TimeoutError:
            print("访问失败")
      return data


def asyn(fun):
    loop = asyncio.get_event_loop()
    tasks = asyncio.ensure_future(fun)
    loop.run_until_complete(tasks)
    # loop.close()
    print('Task ret: {}'.format(tasks.result()))
    return tasks.result()</span>
请求参数的处理
<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>def paramsFilter(params):
    '''
    请求参数处理
    :param params:
    :return:
    '''

    result = {}
    for wap_key in params:
      for son_key in params:
            if params["error"] == "0" or params["error"] == "1" or params["error"] == "2": # 过滤和处理相关请求参数
                result = changeFormat(params["type"], params["input"])
            break
    return result


def changeFormat(key, param):
    '''
    请求参数类型处理
    :param key:
    :param param:
    :return:
    '''
    param_type = {
      "str": lambda: str(param),
      "int": lambda: int(param),
      "float": lambda: float(param),
      "bool": lambda: bool(param)
    }
    return param_type()

def readReq(param):
    '''
    读取请求的url
    :param param:
    :return:
    '''
    return param.split("|") # 1 用例id,2 用例介绍,3 url

def readParam(param):
    '''
    读取准备的pict参数
    param1:...
    param2:..
    :return: list
    '''
    result = []
    _param2 = ""
    for item in param:
      for key in item:
            tempParam = item.split("&")
            _param = ""
            for tItem in tempParam:
                tiParam = tItem.split("|")
                if len(tiParam) == 5:
                  _param = _param + "," + key + ":error:" + tiParam + ":input:" + tiParam + ":type:" + tiParam[
                        2] + ":" + tiParam + ":" + tiParam
                else:
                  _param = _param + "," + key + ":error:" + tiParam + ":" + tiParam + ":" + tiParam
            _param2 = _param2 + "," + _param
            result.append(key + ":"+_param)
            break
    return result


def readPictParam(paramRequestPath):
    '''
    读取本地e生成好了的接口请求参数
    :param paramRequestPath:已经处理好的pict参数路径
    :return: list
    '''
    result = read(paramRequestPath)
    l_result = []
    if result:
      for info in range(len(result)):
            for item in range(len(result)):
                t_result = result.split(",")
                d_t = {}
                for i in t_result:
                  temp = i.split(":")
                  t = {}
                  t] = temp
                  if len(temp) > 5: #如果大于5，说明全部参数为8：如 ：' {'rep': 'dict', 'type': 'str', 'input': '""', 'error': '2'}
                        t] = temp
                        t] = temp
                        t] = temp
                  else:
                        t] = temp # 参数至少
                  d_t] = t
                l_result.append(d_t)
    return l_result


def pairPatchParam(**kwargs):
    '''
       pict生成请求参数
       :param kwargs:
       params: 请求的参数列表，类型为list
       paramPath: 用例目录
       paramRequestPath： 已生成用例目录
       :return:
       '''

    for item in kwargs["params"]:
      write(kwargs["paramPath"], item)
    os.popen("pict " + kwargs["paramPath"] + ">" + kwargs["paramRequestPath"])
    time.sleep(1)</span>

实例-登录
配置init.yaml
<font face="宋体"><pre style='border-width: 1px 0px; margin: 0px -15px 20px; padding: 5px 15px; border-radius: 0px; color: rgb(68, 68, 68); text-transform: none; line-height: 1.4285; text-indent: 0px; letter-spacing: 0.45px; overflow: auto; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; border-top-color: rgb(240, 240, 240); border-bottom-color: rgb(240, 240, 240); border-top-style: solid; border-bottom-style: solid; display: block; word-break: break-all; word-wrap: break-word; box-sizing: border-box; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px; border-image: initial;' class="highlight plaintext"><code style='margin: 2px; padding: 0px !important; border-radius: 3px; border: currentColor !important; color: rgb(68, 68, 68) !important; line-height: 18px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px !important; display: block; white-space: pre-wrap; word-break: break-all; box-sizing: border-box; background-color: rgb(249, 249, 249) !important;'>title: XXXX接口测试
host: baidu.com
port: 8443
protocol: https://
header: {account": "XX", "Content-Type": "application/json; charset=UTF-8","secrectKey": "XXX=","appID": "XX"}
</code></pre><p style='margin: 0px 0px 20px; color: rgb(34, 37, 39); text-transform: none; line-height: 26px; text-indent: 0px; letter-spacing: 0.45px; font-family: Helvetica, Arial, "PingFang SC", "Noto Sans", Roboto, "Microsoft Yahei", sans-serif; font-size: 15px; font-style: normal; font-weight: normal; word-spacing: 0px; white-space: normal; box-sizing: border-box; orphans: 2; widows: 2; background-color: rgb(255, 255, 255); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'><strong style="color: rgb(0, 0, 0); font-weight: 700; box-sizing: border-box;"></strong> </p></font>配置用例yaml


<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>req: 1001|登录|/XX/login|POST
param:
- name: 0|swx458348|str|rep|dict&1|swx4583481|str|rep|dic&3|rep|dict
- XX: 0|fnNoaWt1bjE5ODk|str|rep|dict&1|fnNoaWt1bjE5ODk1|str|rep|dic&3|rep|dict

#error: 0正常，1错误的值，2类型错误,3不传字段，4后面再扩展如最大，最小
#rep：后面是检查点，支持
#{} ,对应key为Dict
#{[]},对应key为DictList
#{[{},{}]} 对应key为DictListDict
</span>


小爸爸 发表于 2017-6-28 11:32:13

本帖最后由 小爸爸 于 2017-6-28 11:35 编辑

PageObject

<p><span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>class Login:
    '''
    kwargs:
    path: 用例文件目录
    initPath： 请求头部目录
    '''

    def __init__(self, **kwargs):
      self.path = kwargs["path"]# 用例yaml目录
      self.param = getYam(self.path)["param"]# 请求参数
      self.req = getYam(self.path)["req"]# 请求url
      self.readParam = readParam(self.param)# 读取并处理请求参数
      pairPatchParam(params=self.readParam, paramPath=PATH("../Log/param.log"),
                     paramRequestPath=PATH("../Log/paramRequest.log"))# pict生成参数
      self.getParam = readPictParam(paramRequestPath=PATH("../Log/paramRequest.log"))# 得到pict生成的参数
      self.readReq = readReq(self.req)# 0 用例id,1 用例介绍,2 url,3 mehtod
      print(self.readReq)
      self.head = requestHead(kwargs["initPath"]) # initPath 请求头准备
      print(self.head)
      # self.head = requestHead(PATH("../yaml/init.yaml"))# protocol ,header,port,host,title

    '''
    发请求
    '''

    def operate(self):
      for item in self.getParam:
            param = paramsFilter(item) # 过滤接口,如果有其他加密，可以自行扩展
            f = request(header=self.head["header"], host=self.head["host"], protocol=self.head["protocol"], port=self.head["port"])
            if self.readReq == "POST":
                BaseAsy.asyn(f.post(self.readReq, param=param))
            else:
                BaseAsy.asyn(f.get(self.readReq, param=param))
</span>
</p>test
<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>from PageObject.PageLogin import Login

PATH = lambda p: os.path.abspath(
    os.path.join(os.path.dirname(__file__), p)
)


class LoginTest(unittest.TestCase):

    def testLogin(self):
      login = Login(path=PATH("../yaml/login.yaml"), initPath=PATH("../yaml/init.yaml"))
      login.operate()
)
</span>

代码入口实例
<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>from Base.BaseRunner import ParametrizedTestCase
from test.TestLogin import LoginTest

def runnerCase():
    starttime = datetime.now()
    suite = unittest.TestSuite()
    suite.addTest(ParametrizedTestCase.parametrize(LoginTest))
    unittest.TextTestRunner(verbosity=2).run(suite)
    endtime = datetime.now()
if __name__ == '__main__':
    runnerCase()</span>
结果执行过程
<span style='color: rgb(68, 68, 68); text-transform: none; text-indent: 0px; letter-spacing: 0.45px; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; font-size: 12px; font-style: normal; font-weight: normal; word-spacing: 0px; float: none; display: inline !important; white-space: pre-wrap; orphans: 2; widows: 2; background-color: rgb(249, 249, 249); font-variant-ligatures: normal; font-variant-caps: normal; -webkit-text-stroke-width: 0px;'>https://XXX post接口参数为:...
Task ret: {'status_code': 409}
https://XXX post接口参数为:{}
{'status_code': 400}
Task ret: {'status_code': 400}
https://XXX/login post接口参数为:....
{'status_code': 200, 'resultCode': 0, 'info': 'Success', '...
https:/...r/login post接口参数为:{'name': 'XXX', 'pwd': 'XXX'}
{'status_code': 409}
....
[[{'param': {'XX': 'AAAAAAAAAAAA'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa', 'XX': 'AAAAAAAAAAAA'}, 'result': {'data': {'resultList': {'token': '5A2EFCE9F0C12EF99D8D18D930D7B71E:8CD9AFC8F3FB5B6B072FAE403A522D9040769B62FB557F90786DF4FFC1AF6356BA269B6E8905197B5B55359D0066146A', 'result': 'success'}}, 'resultCode': 0, 'info': 'Success', 'status_code': 200}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa', 'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1', 'XX': 'AAAAAAAAAAAA'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa'}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1'}, 'result': {'status_code': 400}, 'url': '/XX/login', 'method': 'POST'}, {'param': {'name': 'aa1', 'XX': 'AAAAAAAAAAAA1'}, 'result': {'status_code': 409}, 'url': '/XX/login', 'method': 'POST'}]]



......
</span>

草帽路飞UU 发表于 2017-6-28 17:32:27

楼主你这是针对参数少的，像我们公司接口都是20多个参数，难道也这样去排列组合？那接口多的话是不是太耗时间

乐哈哈yoyo 发表于 2017-6-28 17:32:57

有个疑问, 一组要求顺序POST的请求, 可以用 asyncio 吗
页: [1]
查看完整版本: PICT 组合 +PageObject+Python asyncio 接口测试