流量擷取與入庫

我們需要先定義和初始化流量擷取後儲存的表結構，這裡我們採用peewee這個ORM庫，進行操作。
更多操作可參考：
http://docs.peewee-orm.com/en/latest/peewee/quickstart.html

from peewee import *
import datetime
//可以考慮通過 docker 快速搭建 Mysql 服務
db = MySQLDatabase("mitmproxy", host="172.17.0.2", port=3306, user="root", passwd="xxxx")
db.connect()

class HttpRecords(Model):
    
 
//定義了三個欄位，id, http 請求記錄，時間戳
    id = BigAutoField(primary_key=True)
    httprecord = TextField()
    timestamps = DateTimeField(default=datetime.datetime.utcnow)

    class Meta:
        database = db

接下來編寫我們的外掛指令碼，用於攔截請求，並且寫入到我們定義好的資料庫表中，如下：

//關注介面（get 請求）
get_url = "https://frodo.douban.com/api/v2/user/91807076/following
 
"

//發表說說的介面(post 請求)
post_url = "https://frodo.douban.com/api/v2/status/create_status?loc_id=118282"
httprecord = {}
url_list = [get_url, post_url]

def parser_data(query):
    data = {}
    for key, value in query.items():
        data[key] = value
    return data

class HttpRecord:
    @concurrent
    def
 
 request(self, flow: http.HTTPFlow):
        //這裡可以根據自身業務需求，攔截特定域名下的請求（這裡為了方便演示，特地指定了兩個介面地址）
        if flow.request.pretty_url.startswith(get_url):
            httprecord['method'] = flow.request.method
            httprecord['scheme'] = flow.request.scheme
            httprecord['url'] = flow.request.pretty_url
            httprecord['request_headers'] = {}
            for item in flow.request.headers:
                httprecord['request_headers'][item] = flow.request.headers[item]
            httprecord['get_data'] = parser_data(flow.request.query)
            httprecord['post_data'] = parser_data(flow.request.urlencoded_form)

    @concurrent
    def response(self, flow: http.HTTPFlow):
        if flow.request.pretty_url.startswith(get_url):
            httprecord['status_code'] = flow.response.status_code
            httprecord['response_headers'] = {}
            for item in flow.response.headers:
                httprecord['response_headers'][item] = flow.response.headers[item]
            httprecord['response_content'] = flow.response.get_text()

            # 插入資料庫
            record = HttpRecords(httprecord=httprecord)
            record.save()

addons = [
    HttpRecord()
]

檢查資料庫表，看是否插入成功，如下：

流量回放進行介面測試

從資料庫中查詢請求記錄，並按不同請求方法，e.g.get/post進行分類；
通過request網路請求庫，進行重新回放請求介面；
引入Pytest測試框架，加入斷言，進行組織測試用例的執行，具體程式碼可參考如下：

import demjson
import requests
from replay.httpmodel import HttpRecords
import pytest

class TestInterface:
    get_http = []
    post_http = []
    def setup_class(self):
        # 從資料庫獲取流量記錄(前置處理操作)
        self.httprecords = HttpRecords.select()
        for item in self.httprecords:
            data = demjson.decode(item.httprecord)
            if data['method'] == "GET":
                self.get_http.append(data)
            elif data['method'] == "POST":
                self.post_http.append(data)
            else:
                ...
        # 初始化請求 session
        self.session = requests.session()

    def testReplayGet(self):
        """
         測試回放 Get 請求
        """
        for i in range(len(self.get_http)):
            res = self.session.get(url=self.get_http[i]['url'], headers=self.get_http[i]['request_headers'], data=self.get_http[i]['get_data'])
            //這裡主要是斷言了響應狀態碼，在實際業務中，我們還需要斷言返回格式及校驗核心欄位。
            assert res.status_code == 200

    def testRelayPost(self):
        """
        測試回放 Post 請求
        """
        for i in range(len(self.post_http)):
            res = self.session.post(url=self.post_http[i]['url'], headers=self.post_http[i]['request_headers'], data=self.post_http[i]['post_data'])
            assert res.status_code == 200

if __name__ == '__main__':
    pytest.main(["--html=report.html --self-contained-html", "interfacereplay.py"])

檢視測試報告
get和post請求分別回放測試成功：