自定义关键字

自定义关键字

按照自定义关键字规范，用户可以编写自定义关键字，然后用于用例中。

进入自定义关键字的菜单路径是 数据服务->KEYWORD/PYTHON模式->新增关键字

注：平台不允许用户写代码时使用import引入包，所有有import引用包的需求需要联系管理员添加。

默认引入的包

编写自定义关键字时，默认引入了如下包：

  1. #字符串处理相关的库
  2. import json,re,jsonpath,hashlib,ast
  3. from urllib import parse
  4. from functools import reduce
  5. from bs4 import BeautifulSoup
  6. #时间处理相关的库
  7. import datetime,time,calendar
  8. #数学处理
  9. import math,cmath,decimal
  10. #web请求相关
  11. import requests
  12. #异常处理
  13. import traceback
  14. #数据库相关
  15. import pymysql,redis
  16. from kafka import KafkaProducer
  17. from kafka import KafkaConsumer
  18. from kafka.errors import KafkaError

自定义关键字规范

编写自定义关键字，需要遵循如下规范：

  1. @keyword()
  2. @catch_exception
  3. def DEMO_SELF_KEYWORD(value,context,strTobeProcessed = ""):
  4.     #你调用关键字时传入的参数列表，
  5.     #例如DEMO_SELF_KEYWORD(a,b,c)，处理后的paramList就是列表['a','b','c']
  6.     paramList = getParamList(strTobeProcessed, context)
  7.     #例如DEMO_SELF_KEYWORD(p1,p2,k1 = v1,k2 = v2),处理后的paramList就是列表['p1','p2'],paramDict就是字典{'k1':'v1','k2':'v2'}
  8.     paramList,paramDict = getParamListAndKWParam(strTobeProcessed, context)
  10.     #你调用的关键字处理后的值，未分割
  11.     #例如DEMO_SELF_KEYWORD($VAR[a],b,c)，假设变量a=1，处理后的strTobeProcessed就是字符串1,b,c
  12.     strTobeProcessed = core.processor.KP.KP.getProcessedValue(strTobeProcessed, context)
  14.     #你的代码在这里面写↓
  15.     #return 是你想要的返回值，必须有，按照自己的要求写返回值类型是str
  16.     return int(paramList[0])+int(paramList[1])+int(paramList[2])

说明：

1、开头必须使用装饰器@keyword @catch_exception。
2、函数定义必须符合规范。def DEMO_SELF_KEYWORD(value,context,strTobeProcessed = ""):。
3、获取参数列表方式：paramList = getParamList(strTobeProcessed, context)
4、获取参数列表的字典参数的方式：paramList,paramDict = getParamListAndKWParam(strTobeProcessed, context)
5、处理所有入参，即整个参数只有1个参数时：strTobeProcessed = core.processor.KP.KP.getProcessedValue(strTobeProcessed, context)
6、return 必须有。
7、关键字函数的参数value,context,strTobeProcessed = ""的说明：
value：无需关心，装饰器keyword会处理。 context：上下文对象，对象中会有各种属性和函数，详见context对象说明。 strTobeProcessed：要处理的子串，也就是关键字调用时的参数部分。

自定义关键字示例

登录LOGIN

假设需要实现一个HTTP接口的登录关键字，传入用户名密码，请求登录接口/user/login，然后进行登录。

调用示例：LOGIN(zhangsan,123456);

知识点：

1、生成参数列表list。

2、使用上下文的session。

3、使用上下文的datalist传递登录状态，在一个会话中实现一次登录后续免登陆操作。

  1. @keyword()
  2. @catch_exception
  3. def LOGIN(value,context,strTobeProcessed = ""):
  4.     #LOGIN(username,password),执行下面语句后，paramList = ["username","password"]
  5.     paramList = getParamList(strTobeProcessed, context)
  6.     retmsg = "" # 要返回的信息
  7.     if len(paramList) != 2:
  8.         # 参数长度不符合要求，报错。
  9.         retmsg = "<ERROR: 参数错误，长度必须是2>"
  10.         context.setERROR(retmsg)
  11.         return retmsg
  13.     #参数校验通过后，给username和password赋值。
  14.     username = paramList[0]
  15.     password = paramList[1]
  16.     host = context.host
  18.     # 进行免登陆判断，当执行任务或者业务流时，可以不重复登录。
  19.     # context.context_data_list 是上下文对象中的上下文传递数据列表。可以存储过程数据。
  20.     isContinue = False # 是否继续登录，如果用户名密码主机都一致，就继续，不重新登录。
  21.     if len(context.context_data_list) == 3:
  22.         if username != context.context_data_list[0].strip() or password != context.context_data_list[1].strip() or host != context.context_data_list[1].strip():
  23.             isContinue = True
  24.     else:
  25.         isContinue = True
  27.     if isContinue:
  28.         # 如果继续，重置上下文数据列表。
  29.         context.context_data_list = []
  30.     else:
  31.         return "【已登录】用户%s-Host[%s]，不再次执行登录！" % (username,host)
  33.     # 使用上下文的session进行登录操作。
  34.     context.current_session = requests.session() # 重置session，然后使用新session重新登录。
  35.     resp_obj = context.current_session.post(host+"/user/login",data="username=%s&password=%s" % (username,password))
  36.     if "success" in resp_obj.text:
  37.         # 登录成功后，添加数据到 上下文数据列表 context.context_data_list
  38.         context.context_data_list.append(username)
  39.         context.context_data_list.append(password)
  40.         context.context_data_list.append(host)
  41.         return "登录成功！"
  42.     else:
  43.         # 登录失败，重置session。
  44.         context.current_session = requests.session() # 重置session，然后使用新session重新登录。
  45.         return "登录失败！"

使用MD5生成签名 SIGN

假设后端需要使用md5对密钥和当前时间戳进行加密的数据进行验签。

调用示例：signmd5str = SIGN(secretkey,timestamp);

  1. @keyword()
  2. @catch_exception
  3. def SIGN(value,context,strTobeProcessed = ""):
  4.     # 将参数处理一下，生成最终字符串
  5.     paramList = getParamList(strTobeProcessed, context)
  6.     retmsg = "" # 要返回的信息
  7.     if len(paramList) != 2:
  8.         # 参数长度不符合要求，报错。
  9.         retmsg = "<ERROR: 参数错误，长度必须是2>"
  10.         context.setERROR(retmsg)
  11.         return retmsg
  12.     #参数校验通过后，给username和password赋值。
  13.     secretKey = paramList[0]
  14.     timestamp = paramList[1]
  15.     #对key进行排序
  16.     md = hashlib.md5()  # 创建md5对象
  17.     md.update(( timestamp + secretKey).encode(encoding='utf-8'))
  18.     signature=md.hexdigest()
  19.     #你的代码在这里面写↓
  20.     #return 是你想要的返回值，必须有，按照自己的要求写返回值类型是str
  21.     return signature

results matching ""

No results matching ""