Httprunner集成Sql验证
-
开发功能点：
1、sql语句可参数化
2、sql结果支持JsonPath提取
3、查询结果收集到httprunner Summary信息中
4、定制场景请求前与请求后查询（如：充值接口） - 代码
- 继承Runner类，重写_run_test方法
from httprunner.runner import Runner
from httprunner import exceptions, logger, response, utils
from httprunner.validator import Validator
from utils.httpManage.db import DbMysql
from httprunner.client import HttpSession
class ARunner(Runner):
""" Running testcases. """
def __init__(self, config, http_client_session=None):
"""
config = {
"name": "XXXX",
"base_url": "http://127.0.0.1",
"verify": False,
"db_connect": { # db_mysql connect
"host": "",
"port": "",
"user": "",
"pwd": "",
"database": "",
}
}
"""
super(ARunner, self).__init__(config, http_client_session)
self.config_db_connect = config.get("db_connect", {}) # config增加sql连接信息
self.config_db_connect.update({"session_context": self.session_context})
self.sql_results = {} # sql查询结果
def do_query_actions(self, actions, a_type):
""" 增加sql查询操作方法
Args:
actions (list):
[{"variable_name": '', "sql": "", "json_path":""}]
a_type (str): setup_db/teardown_db
"""
logger.log_debug("\n==================query {} details ==================\n".format(a_type))
with DbMysql(**self.config_db_connect) as db:
db.do_find(actions)
self.sql_results[a_type] = db.sql_results
def _run_test(self, test_dict):
""" run single teststep.
Args:
test_dict (dict): teststep info
{
"name": "teststep description",
"skip": "skip this test unconditionally",
"times": 3,
"variables": [], # optional, override
"request": {
"url": "http://127.0.0.1:5000/api/users/1000",
"method": "POST",
"headers": {
"Content-Type": "application/json",
"authorization": "$authorization",
"random": "$random"
},
"json": {"name": "user", "password": "123456"}
},
"extract": {}, # optional
"validate": [], # optional
"setup_hooks": [], # optional
"teardown_hooks": [] # optional
"setup_dbs": [
{"variable_name": '', "sql": "", "json_path":""}
...
],
"teardown_dbs":[
{"variable_name": '', "sql": "", "json_path": ""}
...
]
}
Raises:
exceptions.ParamsError
exceptions.ValidationFailure
exceptions.ExtractFailure
"""
# clear meta data first to ensure independence for each test
self.__clear_test_data()
# check skip
self._handle_skip_feature(test_dict)
# prepare
test_dict = utils.lower_test_dict_keys(test_dict)
test_variables = test_dict.get("variables", {})
self.session_context.init_test_variables(test_variables)
# teststep name
test_name = self.session_context.eval_content(test_dict.get("name", ""))
# parse test request
raw_request = test_dict.get('request', {})
parsed_test_request = self.session_context.eval_content(raw_request)
self.session_context.update_test_variables("request", parsed_test_request)
# prepend url with base_url unless it's already an absolute URL
url = parsed_test_request.pop('url')
base_url = self.session_context.eval_content(test_dict.get("base_url", ""))
parsed_url = utils.build_url(base_url, url)
# setup hooks
setup_hooks = test_dict.get("setup_hooks", [])
if setup_hooks:
self.do_hook_actions(setup_hooks, "setup")
# setup dbs 新增请求前sql查询
setup_dbs = test_dict.get("setup_dbs", [])
if setup_dbs:
self.do_query_actions(teardown_dbs, "setup_db")
try:
method = parsed_test_request.pop('method')
parsed_test_request.setdefault("verify", self.verify)
group_name = parsed_test_request.pop("group", None)
except KeyError:
raise exceptions.ParamsError("URL or METHOD missed!")
# TODO: move method validation to json schema
valid_methods = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
if method.upper() not in valid_methods:
err_msg = u"Invalid HTTP method! => {}\n".format(method)
err_msg += "Available HTTP methods: {}".format("/".join(valid_methods))
logger.log_error(err_msg)
raise exceptions.ParamsError(err_msg)
logger.log_info("{method} {url}".format(method=method, url=parsed_url))
logger.log_debug(
"request kwargs(raw): {kwargs}".format(kwargs=parsed_test_request))
# request
resp = self.http_client_session.request(
method,
parsed_url,
name=(group_name or test_name),
**parsed_test_request
)
resp_obj = response.ResponseObject(resp)
# teardown hooks
teardown_hooks = test_dict.get("teardown_hooks", [])
if teardown_hooks:
self.session_context.update_test_variables("response", resp_obj)
self.do_hook_actions(teardown_hooks, "teardown")
self.http_client_session.update_last_req_resp_record(resp_obj)
# teardown dbs 新增请求后查询sql
teardown_dbs = test_dict.get("teardown_dbs", [])
if teardown_dbs:
self.do_query_actions(teardown_dbs, "teardown_db")
# extract
extractors = test_dict.get("extract", {})
extracted_variables_mapping = resp_obj.extract_response(extractors)
self.session_context.update_session_variables(extracted_variables_mapping)
# validate
validators = test_dict.get("validate") or test_dict.get("validators") or []
validate_script = test_dict.get("validate_script", [])
if validate_script:
validators.append({
"type": "python_script",
"script": validate_script
})
validator = Validator(self.session_context, resp_obj)
try:
validator.validate(validators)
except (exceptions.ParamsError,
exceptions.ValidationFailure, exceptions.ExtractFailure):
err_msg = "{} DETAILED REQUEST & RESPONSE {}\n".format("*" * 32, "*" * 32)
# log request
err_msg += "====== request details ======\n"
err_msg += "url: {}\n".format(parsed_url)
err_msg += "method: {}\n".format(method)
err_msg += "headers: {}\n".format(parsed_test_request.pop("headers", {}))
for k, v in parsed_test_request.items():
v = utils.omit_long_data(v)
err_msg += "{}: {}\n".format(k, repr(v))
err_msg += "\n"
# log response
err_msg += "====== response details ======\n"
err_msg += "status_code: {}\n".format(resp_obj.status_code)
err_msg += "headers: {}\n".format(resp_obj.headers)
err_msg += "body: {}\n".format(repr(resp_obj.text))
logger.log_error(err_msg)
raise
finally:
# get request/response data and validate results
self.meta_datas = getattr(self.http_client_session, "meta_data", {})
self.meta_datas["validators"] = validator.validation_results
self.meta_datas["sql_results"] = self.sql_results
- 定制sql查询
pip install pymysql
# -------------------------------------
# Author: ALan
# Date: 2022/4/19
# Time: 13:36
# -------------------------------------
import pymysql
import jsonpath
from pymysql import OperationalError
from httprunner import logger, exceptions
class DbMysql(object):
def __init__(self, host, port, user, pwd, database, session_context=None):
# 建立连接
try:
self.conn = pymysql.connect(host=host, # 数据库地址
port=int(port), # 端口号
user=user, # sql账号
password=pwd, # sql密码
database=database, # 数据库名
charset='utf8',
cursorclass=pymysql.cursors.DictCursor)
except OperationalError as err_msg:
logger.log_error(str(err_msg))
else:
# 创建游标
self.cur = self.conn.cursor()
self.session_context = session_context
self.sql_results = []
def __enter__(self):
"""查询前操作"""
self.conn.commit() # 查询前，提交事务
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""执行后关闭连接操作"""
self.cur.close() # 关闭游标
self.conn.close() # 关闭连接
def find_all_result(self, sql):
"""查询sql语句所有结果"""
self.cur.execute(sql) # 执行sql语句
return self.cur.fetchall()
def do_find(self, actions):
"""
查询操作
:param actions: [{"variable_name": '', "sql": "", "json_path":""}]
:return:
"""
self.sql_results = []
def log_print(req_resp_dict):
msg = "\n"
for key, value in req_resp_dict.items():
msg += "{:<16} : {}\n".format(key, repr(value))
logger.log_debug(msg)
for action in actions:
_sql = self.__eval_sql(action.get("sql", "")) # 替换sql语句
json_path = action.get("json_path", "")
sql_value = self._extract_field_with_jsonpath(self.find_all_result(_sql), json_path) # 提取查询数据
sql_variables_mapping = {action.get("variable_name"): sql_value}
self.session_context.update_session_variables(sql_variables_mapping) # 更新结果变量
sql_dict = {
"sql": _sql,
"json_path": json_path,
"sql_value": sql_value
}
log_print(sql_dict)
self.sql_results.append(sql_dict)
def __eval_sql(self, sql_item):
""" evaluate sql item. """
sql_value = self.session_context.eval_content(sql_item)
return sql_value
def _extract_field_with_jsonpath(self, data, field):
"""
:param data: 数据源
:param field: Jsonpath
expression, e.g.
1)$.code
2) $..items. *.id
"""
if field.startswith("$"):
result = jsonpath.jsonpath(data, field)
if result:
return result[0] # 去除列表
else:
raise exceptions.ExtractFailure("\tjsonpath {} get nothing\n".format(field))
else:
return field
- 至此已实现了基础功能点。最终是要完成Sql结果与实际数据校验，那么
Validator类需要重写么？
因原httprunner源码中已实现了Validator，可以不需要改动，仅需在api文件_add_tests方法中把Runner类改成自定义ARunner类就行。但是以上代码遗留了一个问题是在用例结构体validate中定义($参数名），必须在变量variables中定义该参数名称否则运行代码会报找不到参数错误
例：
image.png
-
完整用例结构体
image.png -
日志效果图
image.png
image.png
