持续创造,加快成长!这是我参加「日新计划 6 月更文挑战」的第15天,点击检查活动详情
咱们好~我是
米洛
!
我正在从0到1打造一个开源的接口测试渠道, 也在编写一套与之对应的教程
,期望咱们多多支撑。
欢迎重视我的公众号米洛的测开日记
,获取最新文章教程!
回忆
上一节咱们讲述了用例录制
相关内容,但是有一些websocket的改造和http恳求成果的转化没有讲,这一节咱们就来完结他们。
Websocket改造
websocket改造分为2部分:
- 后端部分要支撑发送录制好的http恳求数据,其实也便是json数据,之前已经支撑,咱们略微改动下即可
- 前端部分需求监听新的类型,收到后端发来的音讯,依据音讯type做出不同的操作
先来看后端部分:
其实咱们只需求在websocketManager加入send_data办法即可,也便是能自由地依据用户id发送音讯。
再来看看前端部分:
略微有点改变,依据type为2,判别是录制成功
的音讯,咱们走存储录制音讯的逻辑。
其实readRecord便是往录制列表里边刺进数据
。
Redis部分
@staticmethod
@awaitable
def get_address_record(address: str):
"""
获取ip是否已经敞开录制
:param address:
:return:
"""
key = RedisHelper.get_key(f"record:ip:{address}")
return RedisHelper.pity_redis_client.get(key)
@staticmethod
@awaitable
def cache_record(address: str, request):
"""
:param address:
:param request:
:return:
"""
key = RedisHelper.get_key(f"record:{address}:requests")
RedisHelper.pity_redis_client.rpush(key, request)
ttl = RedisHelper.pity_redis_client.ttl(key)
if ttl < 0:
RedisHelper.pity_redis_client.expire(key, 3600)
@staticmethod
@awaitable
def set_address_record(user_id: int, address: str, regex: str):
"""
设置录制状况
:param user_id:
:param address:
:param regex: 录制的url正则
:return:
"""
# 默认录制1小时
value = json.dumps({"user_id": user_id, "regex": regex}, ensure_ascii=False)
RedisHelper.pity_redis_client.set(RedisHelper.get_key(f"record:ip:{address}"), value, ex=3600)
# 清楚上次录制数据
RedisHelper.pity_redis_client.delete(RedisHelper.get_key(f"record:{address}:requests"))
@staticmethod
@awaitable
def remove_address_record(address: str):
"""
中止录制任务
:param address:
:return:
"""
return RedisHelper.pity_redis_client.delete(RedisHelper.get_key(f"record:ip:{address}"))
主要由这4个办法组成:
- 获取该ip录制的状况,是敞开还是封闭
- 缓存录制成果(上一节已经介绍过)
- 设置录制状况,1小时后过期,也就贴合之前说的,1小时后任务主动中止
- 封闭录制,直接删除key,提前中止录制
所以对应页面上咱们会有开始录制
,中止录制
等操作按钮。
HTTP恳求解析
这块内容其实不难,便是整合mitmproxy的flow对象里边有的跟http恳求相关的内容,我这儿只选取了一些用得到的部分,比方headers/body/cookies等等。
__author__ = "woody"
import json
from typing import TypeVar
import pydantic
from loguru import logger
"""
translate mitmproxy request and response data
"""
body = TypeVar("body", bytes, str)
class RequestInfo(pydantic.BaseModel):
url: str = None
body: body = None
request_method: str = None
request_data: str = None
request_headers: str = None
response_headers: str = None
cookies: str = None
request_cookies: dict = None
response_content: str = None
status_code: int = None
def __init__(self, flow):
super().__init__()
self.status_code = flow.response.status_code
self.url = flow.request.url
self.request_method = flow.request.method
self.request_headers = json.dumps(dict(flow.request.headers), indent=4, ensure_ascii=False)
self.response_headers = json.dumps(dict(flow.response.headers), indent=4, ensure_ascii=False)
self.response_content = self.get_response(flow.response)
self.body = self.get_body(flow.request)
self.cookies = json.dumps(dict(flow.response.cookies), indent=4, ensure_ascii=False)
self.request_cookies = dict(flow.request.cookies)
@classmethod
def translate_json(cls, text):
try:
return json.dumps(json.loads(text), indent=4, ensure_ascii=False)
except Exception as e:
logger.bind(name=None).warning(f"解析json格局恳求失利: {e}")
return text
@classmethod
def get_response(cls, response):
content_type = response.headers.get("Content-Type")
if "json" in content_type.lower():
return cls.translate_json(response.text)
if "text" in content_type.lower() or "xml" in content_type.lower():
return response.text
return response.data.decode('utf-8')
@classmethod
def get_body(cls, request):
if len(request.content) == 0:
return None
content_type = request.headers.get("Content-Type")
if "json" in content_type.lower():
return cls.translate_json(request.text)
if "text" in content_type.lower() or "xml" in content_type.lower():
return request.text
return request.data.decode('utf-8')
def dumps(self):
return json.dumps(self.dict(), ensure_ascii=False)
值得注意的是,dumps办法,是因为咱们把requestInfo对象存入redis的时候需求序列化,因为redis不允许存入Python数据,需求转化为bytes或许str。
main.py敞开proxy
最后依据proxy的开关判别,是否敞开装备,因为proxy也是一个服务,咱们单独用一个线程发动它(create_task,而且不等候)
前端部分
其实比较干燥,便是一个很简单的表格
,加上对应的url输入框和详细的按钮:
后续咱们需求能够勾选
录制到的恳求,而且做一些智能操作,主动提取里边的参数数据,最后完结用例的生成操作。
其间智能提取参数是比较通用的,需求兼容其他比方har文件格局的数据,最后怎样导入到pity成为pity的一员,这个笔者还在谋划。
今日的内容就到这儿了,感谢咱们的收看,再会~