干货：使用Fastapi开发自己的Mock server(附源码)

最近几天抽空看了下Fastapi的文档，顺便使用它开发了个HTTP(s)协议的Mock server。以后有需要用到挡板的地方，就可以直接使用啦。

项目一级目录：

logs用来存放日志文件，日切

main：启动入口程序，直接运行

requirements.txt：依赖包

common中存放日志模块，核心模块等公用部分。

apis用来存放各个项目的挡板服务，各目录以项目为单位分开；每个项目目录下固定有三个文件夹：

api_operation：存放接收post、get等请求并返回Mock响应的代码。

config：存放Mock的匹配规则

responce_file：存放Mock响应的文件，json、txt.

注意：config下的配置文件名与responce_file要一一对应。

api_operation的开发

import timefrom fastapi import APIRouter, Path, Queryfrom starlette.responses import FileResponsefrom starlette.requests import Requestfrom common.mock_responce import MockResponcefrom common.log_config import getlogerrouter = APIRouter()logger = getloger(name)@router.post("/v1/{url_path}")async def json_res(request: Request, url_path=Path(…, title=“The ID of the item to get”)): print(request.url) # query_args = request.query_params # logger.info(‘query_args:’, query_args) # logger.info(query_args.get(‘entInfo’)) # form_args = await request.form() # print(‘form_args:’, form_args) # print(form_args.get(‘entInfo’)) # byte_body = await request.body() # try: # import json # b_body = json.loads(byte_body.decode(‘UTF-8’)) # print(‘byte_body:’, b_body) # print(type(b_body)) # except: # b_body = byte_body.decode(‘UTF-8’) # print(‘byte_body:’, b_body) # print(type(b_body)) json_body = await request.json() return FileResponse(MockResponce(‘test_api’, ‘A001’, json_body).responce_filter())@router.post("/v1/txt")async def txt_res(request: Request): form_args = await request.form() return FileResponse(MockResponce(‘test_api’, ‘A002’, form_args).responce_filter())
通过request参数可以获取请求传进来的所有header、参数等；

参数通常有三种方法传进来：

1.url传参，query_args = request.query_params来获取

2.form-da传参，form_args = await request.form()来获取

3.body传参，如果穿的是json或json格式的字符串，用json_body = await request.json()来获取；如果穿的是二进制或类json字符串，可以使用byte_body = await request.body()来获取。

然后将取得的值传给核心模块即可：

MockResponce(‘test_api’, ‘A001’, json_body).responce_filter()

test_api：挡板项目名，即apis下一级目录的名称。

A001：规则文件名

json_body：从request所取得的参数

技术细节:

1,2两种方式接收到的参数虽然不是字典，但是都可以使用跟字典一样的方式取得参数，例如，url传参如下：

http://127.0.0.1:8000/vba1/vvv/23123?entInfo=平安银行股份有限公司&appid=weq12311

可以使用query_args[‘entInfo’]或者query_args.get(‘entInfo’)来获取entInfo的值。

这样核心模块就可以将取得的entInfo与规则文件进行匹配，并根据匹配的结果获得相应文件返回。

3这种方式接收的参数，大部分为json格式，可以直接转化为字典，但是在复杂嵌套的json中想要取得entInfo的值，就更复杂，Mock server核心部分已经做了封装，会自动处理。少部分body传进来的可能是二进制，需根据实际情况再处理。

服务路由
以上开发完成后，需要将服务添加路由至fastap的app，类似于Flask的Blueprint,具体位置如下图

在这里还可以设置请求路径共同的部分prefix等。

from fastapi import FastAPIfrom apis.test_api.api_operation import
study1,study2from apis.CIF_api.api_operation import
bindcard,evalPhonefrom apis.vba_api.api_operation import vba1,vba2def
create_app(): app = FastAPI(title=‘FastApi Mock Server’,
description=‘这是使用fastapi开发的挡板服务程序，旨在使用简单的操作实现快速可用的挡板服务’,
version=0.1) app.include_router( study1.router,
prefix="/items", tags=[“items”], dependencies=[],
responses={404: {“description”: “Not found”}}, )
app.include_router( study2.router, prefix="/test",
tags=[“test”], dependencies=[], responses={404:
{“description”: “Not found”}}, ) app.include_router(
bindcard.router, prefix="/bindcard", tags=[“bindcard”],
dependencies=[], responses={404: {“description”: “Not found”}},
) app.include_router( evalPhone.router,
prefix="/evalPhone", tags=[“evalPhone”],
dependencies=[], responses={404: {“description”: “Not found”}},
) app.include_router( vba1.router, prefix="/vba1",
tags=[“vba1”], dependencies=[], responses={404:
{“description”: “Not found”}}, ) app.include_router(
vba2.router, prefix="/vba2", tags=[“vba2”],
dependencies=[], responses={404: {“description”: “Not found”}},
) return app

config下规则文件的定义
#入参为空，value:None#vba1即该文件的名字，必须保持一致，纯数字要加引号，不然会被yaml当做intvba1: - res_file: 平安银行股份有限公司.json enable: Y index: 1 delay: 0 condition: - {field: entInfo,rule: is,value: 平安银行股份有限公司} - {field: appid,rule: is,value: weq12313331} - res_file: 查询无记录.json enable: Y index: 2 delay: 0 condition: - {field: entInfo,rule: is,value: 查询无记录} - res_file: 入参为空.json enable: Y index: 3 delay: 0 condition: - {field: entInfo,rule: is,value: None} - res_file: 文本响应.txt enable: Y status: 1 index: 7 delay: 0 condition: - {field: entInfo,rule: regex,value: ‘.+’}
res_file:responce_file对应目录下存放Mock响应的文件名，必须带后缀。

enable：是否启用该规则，Y,y启用，N，n不启用

index：规则序号

delay：响应延时，为空默认为0，例如delay: 5，如果匹配到这条规则，则延时5秒再返回响应。

condition：匹配该规则的条件，如果多个条件为且的关系，所有规则并列放在一起，必须满足全部；如果多条规则为或的关系，则所有规则分开写，命中任意一条即返回。

field：需要用来匹配的关键字，如上则表示需要从请求的参数中拿到entInfo的值

rule：匹配规则，is-完全一样；regex-正则匹配

value：规则的期望值，field中取得的值与value中的值匹配。value: None表示传入的参数为空或未传入该参数。

规则匹配顺序，按照规则文件中写的顺序，从上到下，命中即返回。

技巧：

在规则文件末尾一般放一个全匹配的正则规则兜底，保证任何请求都有响应：

  - res_file: 文本响应.txt    enable: Y    status: 1    index: 7    delay: 0    condition:      - {field: entInfo,rule: regex,value: '.+'}

服务启动

以上都开发好后，使用时，在main.py同级目录的命令行运行：

uvicorn main:app --host 0.0.0.0 --port 8000 --reload

启用HTTPS:

uvicorn main:app --host 0.0.0.0 --port 8000 --reload --ssl-keyfile server.key --ssl-certfile server.crt

如果要启用http，则加入server.crt与server.key启动服务，并将ca.crt追加到客户端linux服务器的授信列表：

cat ca.crt>> /etc/pki/tls/certs/ca-bundle.crt

或者直接运行main.py启动服务

import uvicornfrom fastapi import Depends, FastAPI, Header, HTTPExceptionfrom apis import create_appfrom common.log_config import getlogerapp = create_app()logger = getloger(__name__)@app.on_event("shutdown")def shutdown_event():    print('shut down')@app.on_event("startup")async def startup_event2():    print('startup')if __name__ == "__main__":    uvicorn.run('main:app', host="0.0.0.0", port=8000, reload=True)    # uvicorn.run('main:app', host="0.0.0.0", port=8000, reload=True, ssl_keyfile='server.key', ssl_certfile='server.crt')

Mock server使用

使用时，只需要修改规则yaml文件和responce_file响应文件即可。

例如，对于上面的规则，body给不同的传参效果：

此外，由于fastapi的特点，还可以使用自带的swagger服务查看所有挡板服务：

代码已上传中github:http://github.com/superxuu/fastapi_mock

今天分享就到这里，喜欢的点个关注~~

最后： 给大家推荐一个 q 群：902061117 里面有许多资料共享！资料都是面试时面试官必问的知识点，也包括了很多测试行业常见知识，其中包括了有基础知识、Linux必备、Shell、互联网程序原理、Mysql数据库、抓包工具专题、接口测试工具、测试进阶-Python编程、Web自动化测试、APP自动化测试、接口自动化测试、测试高级持续集成、测试架构开发测试框架、性能测试、安全测试等。

如果对你有一点点帮助，各位的「点赞」就是小编创作的最大动力，我们下篇文章见！

好文推荐

2021软件测试工程师面试题汇总（内含答案）-看完BATJ面试官对你竖起大拇指！

什么样的人适合从事软件测试工作？

软件测试和软件开发哪个发展更好

那个准点下班的人，比我先升职了…