流量回放系统的设计与实现--流量回放模块_流量回放自定义规则比对结果

架构设计

流程回放功能数据流转示意图

 实现效果

录制：

回放：

功能设计

流量回放模块主要包含的功能是：

• 任务信息列表

• 创建回放任务

• 修改任务信息

• 执行任务

• 回放对比结果

• 查看任务详情信息

• 流量用例列表

• 用例详情、编辑

问题与解决方案

1队列中很多流量，如何获取指定的部分流量？

解决方案：通过对mitmproxy的二次开发，我们增加了根据指定列表id的dump接口。前端在列表选择的流量id，直接调用暴露的api接口，遍历获取到的id，自定义过滤获取指定的流量文件，同时解析流量数据并入库保存。此时的流量文件保存路径也会入库，以便于后续回放任务指定流量文件。

2回放对比时，部分请求流量很大，对比起来很慢如何处理？

解决方案：通过全量对比返回数据的形式是不可取的，特别是流量数据大时，很容易造成前端界面的卡死。这里我们通过对返回数据建立hash值的方式，直接对比2次的hash值来获得回放的对比结果，大大提高数据对比模块的性能。

3回放时如何让新的流量与老的录制流量进行关联？

解决方案：此种基于mitmproxy自身流量文件的回放方式，回放时的request流量序列id是不变的，由于在生成流量用例时已经将老的流量返回值入库保存，后续回放时通过中间插件解析并根据id也对新的response流量进行保存，通过唯一的id进行了关联。

4流量回放时如何自定义去修改请求？

解决方案：这个功能平台目前没有开发实现，但是前期已经调研尝试过，可以过开发中间件脚本对流量进行自定义的编辑处理。

核心代码设计

前段对比模块

这里采用的是 react-diff-viewer 库，组件化剥离了数据对比模块。

// DiffView.js
import React, { Component } from 'react';
import ReactDiffViewer, { DiffMethod } from 'react-diff-viewer';
 
import styles from './index.less';
 
export default class Page extends Component {
  render() {
    const { leftTitle, rightTitle, srcResponse, dstResponse } = this.props;
    return (
      <div className={styles.diffContainer}>
        <ReactDiffViewer
          leftTitle={leftTitle}
          rightTitle={rightTitle}
          oldValue={JSON.stringify(srcResponse, null, 2)}
          newValue={JSON.stringify(dstResponse, null, 2)}
          splitView
          // showDiffOnly={false}
          compareMethod={DiffMethod.JSON}
          // renderContent={this.highlightSyntax}
        />
      </div>
    )
  }
}

mitmproxy中的自定义dump接口

这个实际是在mitmweb模式下才会提供的一个接口，通过忽略跨域，暴露出来给前端调用。

# mitmproxy/tools/web/app.py
class FlowDumpByIdHandler(RequestHandler):
    def get(self):
        self.set_header("Content-Disposition", "attachment; filename=flows")
        self.set_header("Content-Type", "application/octet-stream")
 
        save_ids = self.json
 
        bio = BytesIO()
        fw = io.FlowWriter(bio)
        for f in self.view:
            if save_ids['saveIds']:
                if f.id in save_ids['saveIds']:
                    fw.add(f)
 
        self.write(bio.getvalue())
        bio.close()

获取流量response的中间件

核心的是返回值解析部分，这里也可以不做解析就用原生的json，但是最后入库的数据量会很大，影响后续的界面展示性能，可以看到目前主要是解析了xml和json数据，应该能覆盖常见的类型了，对于后续遇到的其它形式的返回值可以自定义扩展解析。

# getReplayResponse.py
import mitmproxy.http
from bs4 import BeautifulSoup
from mitmproxy import ctx
import logging, json, typing, requests, hashlib
 
PARSE_ERROR = object()
 
def parse_json(s: bytes) -> typing.Any:
    try:
        return json.loads(s.decode('utf-8'))
    except ValueError:
        return PARSE_ERROR
 
class MappingAddonConfig:
    HTML_PARSER = "html.parser"
 
class GetReplayResponse:
    def __init__(self):
        self.serverDomain = "{appHost}"
        self.taskId = "{taskId}"
        self.logger = logging.getLogger(self.__class__.__name__)
 
    def uploadToServer(self, requestId, responseData, replayRespHash):
        url = self.serverDomain + '/api/replayManage/getResponseFromProxy'
        data = {{
            'taskId': self.taskId,
            'requestId': requestId,
            'responseData': responseData,
            'replayRespHash': replayRespHash,
        }}
        headers = {{'Content-Type': 'application/json'}}
        res = requests.post(url, data=json.dumps(data), headers=headers)
        resp = res.json()
        if resp['code'] != 0:
            ctx.log.info("upload [ %s ] flow failed " % requestId)
 
    def response(self, flow: mitmproxy.http.HTTPFlow) -> None:
        """If a response is received, check if we should replace some content. """
        try:
            requestId = flow.id
            res = flow.response
            if res is not None:
                encoding = res.headers.get("content-encoding", "utf-8")
                content_type = res.headers.get("content-type", "text/html")
                replayRespHash = hashlib.sha256(res.raw_content).hexdigest()
                responseData = ''
                if "text/html" in content_type and encoding == "utf-8":
                    content = BeautifulSoup(res.content, MappingAddonConfig.HTML_PARSER)
                    responseData = content.encode(encoding)
                elif "json" in content_type:
                    data = parse_json(res.content)
                    if data is not PARSE_ERROR:
                        responseData = data
                    else:
                        self.logger.warning(
                            f"PARSE_ERROR content type '{{content_type}}'")
                else:
                    self.logger.warning(f"Unsupported content type '{{content_type}}' or content encoding '{{encoding}}'")
                self.uploadToServer(requestId, responseData, replayRespHash)
        except KeyError:
            pass
 
 
addons = [
    GetReplayResponse()
]

本系统采用前后端分离架构，本文仅就 流量回放 模块的设计与实现，做分析讲解，其它模块将在后续文章中拆分讲述。

最后：下方这份完整的软件测试视频教程已经整理上传完成，需要的朋友们可以自行领取【保证100%免费】

软件测试面试文档

我们学习必然是为了找到高薪的工作，下面这些面试题是来自阿里、腾讯、字节等一线互联网大厂最新的面试资料，并且有字节大佬给出了权威的解答，刷完这一套面试资料相信大家都能找到满意的工作。