轻松玩转Mock与接口自动化测试

　　mitmproxy 是什么
　　mitmproxy 是一个免费的开源交互式的 HTTPS 代理。mitmproxy 就是用于 MITM 的 proxy，MITM 即中间人攻击（Man-in-the-middle attack），用于中间人攻击的代理首先会向正常的代理一样转发请求，保障服务端与客户端的通信，其次会看看请求或者响应结果信息，记录其截获的数据或篡改数据，引发服务端或客户端特定的行为。 为什么选择 mitmproxy相对于我们常用的 fiddle 代理工具，它是可以跨平台； 相对于跨平台的 charles 代理工具，它是开源免费的； 最重要的一点，也是为什么我们采用 mitmproxy 来进行做 Mock，它支持使用 Python 进行二次开发，这样就可以结合业务进行灵活拓展了。 mitmproxy 构成mitmproxy：是一个控制台工具，允许交互式检查和修改 HTTP 流量 mitmweb：mitmproxy 是基于 Web 的用户界面，它允许交互式检查和修改 HTTP 流量 mitmdump：mitmproxy 的命令行版本。它提供了类似 tcpdump 的功能，可查看、记录和以编程方式转换 HTTP 流量（我们在接下来的场景中使用的就是这个） 环境安装    sudo apt-get install mitmproxy     pip3 install mitmproxy 核心 API 介绍
　　mitmproxy 插件编写主要是基于事件可以在不同层面上分为 5 类：HTTP、TCP、WebSocket、Network、General。
　　其中最常用的就是 HTTP 事件，主要如下：（其中我们在我们接下来的使用场景主要会用到 request、response 这两个事件函数）
　　1. 收到了来自客户端的 HTTP CONNECT 请求
　　在 flow 上设置非 2xx 响应将返回该响应并断开连接。
　　CONNECT 不是常用的 HTTP 请求方法，目的是与服务器建立代理连接，仅是 client 与 proxy 的之间的交流，所以 CONNECT 请求不会触发 request、response 等其他常规的 HTTP 事件。 def http_connect(self, flow: mitmproxy.http.HTTPFlow):
　　2. 来自客户端的 HTTP 请求的头部被成功读取
　　此时 flow 中的 request 的 body 是空的。 def requestheaders(self, flow: mitmproxy.http.HTTPFlow):
　　3. 来自客户端的 HTTP 请求被成功完整读取 def request(self, flow: mitmproxy.http.HTTPFlow):
　　4. 来自服务端的 HTTP 响应的头部被成功读取
　　此时 flow 中的 response 的 body 是空的。 def responseheaders(self, flow: mitmproxy.http.HTTPFlow):
　　5. 来自服务端端的 HTTP 响应被成功完整读取 def response(self, flow: mitmproxy.http.HTTPFlow):
　　发生了一个 HTTP 错误。比如无效的服务端响应、连接断开等。注意与＂有效的 HTTP 错误返回＂不是一回事，后者是一个正确的服务端响应，只是 HTTP code 表示错误而已。 def error(self, flow: mitmproxy.http.HTTPFlow): 关键字段介绍（flow 对象）flow.request.host    请求 host flow.request.method  请求方法 flow.request.scheme  请求协议 flow.request.url     请求 URL 链接 flow.request.query   请求 URL 查询参数 flow.request.path    请求 URL 路径 flow.request.urlencoded_form  请求 POST 参数 flow.response.status_code  HTTP 响应状态码 flow.response.headers    HTTP 响应头信息 flow.response.get_text   HTTP 响应内容 如何编写一个插件
　　插件的作用主要是用于拦截请求，根据自身业务需求，在不同的事件函数中，添加业务逻辑处理代码。
　　插件的编写格式相对比较简单，如果不熟悉如何编写，可参考官方样例：
　　https://github.com/mitmproxy/mitmproxy/blob/master/examples/addons
　　进行模仿，这里摘取其中一个增加头部信息的样例，如下： class AddHeader:     def __init__(self):         self.num = 0  //这里选择所需要处理的事件函数，e.g. 如对 response 进行处理     def response(self, flow):         self.num = self.num + 1         flow.response.headers[＂count＂] = str(self.num)  //这个是固定格式 addons = [     AddHeader() ] Mock 测试场景示例（以豆瓣 App 为例）
　　场景介绍：测试＂我关注的人＂列表，超过一屏，是否能够正常显示。（为了便于快速测试，所以我们采用 Mock 的方式，构造数据）
　　拦截篡改 http 请求的工具类，包含的功能主要有： 拦截请求并请求 header 信息 拦截请求及修改 body 信息 拦截请求方法及修改类型，e.g. get/post 拦截请求及修改 query 参数 拦截响应并修改响应状态码 拦截响应并修改响应头信息 拦截响应并修改响应正文信息 拦截响应并构造响应信息，e.g. 状态码、响应头、响应体 class HTTPRecordModifier:      def __init__(self, flow: http.HTTPFlow):         self.flow = flow     //设置请求头信息     def set_request_header(self, headers):         for header_key, header_value in headers.items():             self.flow.request.headers[header_key] = header_value      //设置请求 body 参数     def set_request_body(self, body):         self.flow.request.content = bytes(body, ＂utf-8＂)      //设置请求方法     def set_request_method(self, method):         self.flow.request.method = method      //设置请求 query 参数     def set_request_query(self, key, value):         self.flow.request.query[key] = value      //设置响应状态码     def set_response_status_code(self, code):         self.flow.response.status_code = code      //设置响应头信息     def set_response_header(self, headers):         for header_key, header_value in headers.items():             self.flow.response.headers[header_key] = header_value       //设置响应体内容     def set_response_body(self, body):         self.flow.response.content = bytes(body, ＂utf-8＂)      //构造响应报文     def create_mocked_response(self, code=200, header={}, body=＂＂):         self.flow.response = http.HTTPResponse.make(code, bytes(body, ＂utf-8＂), header)
　　抓包获取我们需要 Mock 的接口数据返回的格式，如下： {                 ＂loc＂: None,                 ＂kind＂: ＂user＂,                 ＂followed＂: True,                 ＂name＂: ＂啊浦西＂,                 ＂url＂: ＂https://www.douban.com/people/197322324/＂,                 ＂verify_type＂: 0,                 ＂absct＂: ＂＂,                 ＂reg_time＂: ＂2019-05-30 07:32:57＂,                 ＂avatar＂: ＂https://img9.doubanio.com/icon/up197322324-6.jpg＂,                 ＂uri＂: ＂douban://douban.com/user/197322324＂,                 ＂medal_groups＂: [],                 ＂followers_count＂: 3,                 ＂in_blacklist＂: False,                 ＂uid＂: ＂197322324＂,                 ＂is_banned＂: False,                 ＂type＂: ＂user＂,                 ＂id＂: ＂197322324＂,                 ＂remark＂: ＂＂ }
　　编写具体的插件脚本，在响应事件函数内编写我们的逻辑代码： class Test:      def response(self, flow: http.HTTPFlow):         ＂＂＂         Event for handling response before sending it back to the client         ＂＂＂         //这里编写我们的 mock 逻辑代码         breakpoint_url = ＂https://frodo.douban.com/api/v2/user/91807076/following＂         if breakpoint_url in flow.request.pretty_url:             response_content = json.loads(flow.response.content.decode(＂utf-8＂))             response_content[＂total＂] = 20             new_response = HTTPRecordModifier(flow)             userinfo = {                 //这里放置上面抓包获取的用户信息格式                }             for i in range(response_content[＂total＂]):                 //添加 20 个用户信息                 response_content[＂users＂].append(userinfo)             new_response.set_response_body(json.dumps(response_content))  addons = [     Test() ]
　　手机网络连上代理，具体操作可参考：
　　https://www.cnblogs.com/ninanie/p/11340209.html
　　启动 mitmproxy 插件脚本，如下：  mitmdump -s ./mockbusiness.py (插件的脚本名字)
　　进入豆瓣，我关注的人，查看效果图，可以看到我们构造的 20 个用户数据，如下：
　　以上我们就完成了通过编写插件脚本，进行 Mock 的测试场景。 流量截取与入库
　　我们需要先定义和初始化流量截取后存储的表结构，这里我们采用 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 请求分别回放测试成功：