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

大家好~我是米洛！
我正在从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的一员，这个笔者还在筹划。

今天的内容就到这里了，感谢大家的收看，再会~