测试平台系列(133) 用例生成之录制接口请求（下）

持续创造，加快成长！这是我参加「日新计划 6 月更文挑战」的第15天，点击检查活动详情

咱们好~我是米洛！
我正在从0到1打造一个开源的接口测试渠道, 也在编写一套与之对应的教程，期望咱们多多支撑。
欢迎重视我的公众号米洛的测开日记，获取最新文章教程!

回忆

上一节咱们讲述了用例录制相关内容，但是有一些websocket的改造和http恳求成果的转化没有讲，这一节咱们就来完结他们。

Websocket改造

websocket改造分为2部分:

1. 后端部分要支撑发送录制好的http恳求数据，其实也便是json数据，之前已经支撑，咱们略微改动下即可
2. 前端部分需求监听新的类型，收到后端发来的音讯，依据音讯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个办法组成:

1. 获取该ip录制的状况，是敞开还是封闭
2. 缓存录制成果（上一节已经介绍过）
3. 设置录制状况，1小时后过期，也就贴合之前说的，1小时后任务主动中止
4. 封闭录制，直接删除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的一员，这个笔者还在谋划。

今日的内容就到这儿了，感谢咱们的收看，再会~