1. 项目概述为什么我们需要自己造一个Mock工具在前后端分离、微服务架构大行其道的今天接口联调是每个开发者都绕不开的坎。你肯定遇到过这种情况前端页面急着要上线后端接口还在开发中或者依赖的第三方服务不稳定动不动就给你来个超时。这时候一个靠谱的接口Mock工具就成了救命稻草。市面上的Mock工具不少像Postman的Mock Server、Apifox、YApi还有Python里的responses、httpretty这些库用起来都挺方便。那为什么还要自己动手写一个呢原因很简单不够“趁手”。现有的工具要么太重配置繁琐要么太轻功能单一比如只能返回固定数据无法根据请求参数动态响应更别提对请求体做复杂的校验了。特别是当你需要模拟一个逻辑稍微复杂一点的接口比如请求参数A大于10时返回成功否则返回失败或者需要校验JSON请求体里某个嵌套字段的格式时通用工具往往需要你写一大段脚本或者干脆做不到。这就是“手写Python接口Mock工具”的价值所在。我们不是要做一个替代所有现有工具的巨无霸而是要打造一把高度定制化、深度可控的“瑞士军刀”。基于AST抽象语法树来实现听起来有点“硬核”但这恰恰是它的精髓。AST允许我们以代码结构本身为操作对象这意味着我们可以用写Python逻辑的方式去定义Mock的规则——校验请求、生成响应、处理异常所有这些都能用你熟悉的if-else、循环、函数调用来表达灵活性和可编程性直接拉满。这个项目就是带你从零开始理解Mock的核心需求并用AST这把“手术刀”精细地构建一个属于你自己的、功能强大的Mock工具。2. 核心需求与设计思路拆解在动手写代码之前我们必须把需求掰开揉碎了看。一个功能完备的Mock工具远不止是返回一个{code: 200, “data”: “hello”}这么简单。2.1 核心需求解析接口模拟这是基础。工具必须能监听特定端口和路径对HTTP请求GET、POST等做出响应。它需要模拟真实接口的延迟、状态码、响应头和响应体。请求校验这是保证Mock有效性的关键。我们需要能验证请求的方方面面结构校验请求体是否是合法的JSONXML字段校验某个必填字段是否存在字段的数据类型字符串、数字、数组是否正确业务逻辑校验字段A的值是否在某个范围内字段B是否依赖于字段A的值例如创建订单时金额必须大于0。动态响应这是Mock工具智能化的体现。响应不能是死的它应该能根据请求的内容动态变化。参数化响应将请求中的某个值如user_id填充到响应模板中。条件化响应根据不同的请求参数或头部信息返回完全不同的响应内容和状态码。逻辑生成响应数据可以通过一段逻辑代码计算得出比如生成一个随机的订单号或者计算一个基于请求参数的哈希值。2.2 为什么选择基于AST的实现方案面对这些需求常见的实现思路可能是用一堆if语句和正则表达式在请求处理函数里硬编码。但这样代码会迅速变得臃肿且难以维护特别是当Mock规则复杂时。AST方案的优势在于“声明式”与“运行时编译”。声明式配置我们可以让用户用一种接近自然语言或简化Python语法的方式比如一个字典或一段特定格式的字符串来定义Mock规则。例如{ “path”: “/api/user”, “method”: “POST”, “request_check”: “request.json[‘age’] 0 and len(request.json[‘name’]) 1”, “response”: “{‘id’: random.randint(1000, 9999), ‘name’: request.json[‘name’]}” }AST转换与执行上面的request_check和response是字符串。我们的工具核心就是将这些字符串解析成AST然后在处理真实请求的上下文中编译并执行这棵语法树。request.json[‘age’]在AST中就是一个属性访问节点是一个比较操作节点。执行时request会被替换为实际的请求对象。这样做的好处是安全通过限制AST可访问的变量和函数__builtins__我们可以构建一个沙箱环境防止用户定义的Mock规则执行危险代码如os.system(‘rm -rf /’)。灵活理论上用户可以用任何合法的Python表达式来定义规则能力边界非常广。性能AST可以被编译成字节码code object并缓存。对于高频调用的Mock接口首次编译后后续执行就是直接的字节码解释速度很快。设计蓝图我们的工具将分为三层。配置层负责读取和解析用户定义的规则文件YAML/JSON。核心引擎层包含AST解析器、安全沙箱构造器、规则编译器和执行器。这是最核心的部分。HTTP服务层一个轻量的Web服务器如使用aiohttp或Flask接收请求将请求对象传递给核心引擎匹配并执行规则最后返回引擎生成的响应。3. 核心引擎实现AST的解析、沙箱与执行这是整个项目最硬核、也最有趣的部分。我们将一步步构建核心引擎。3.1 构建安全的AST执行沙箱直接执行来自用户的字符串代码是极度危险的。我们必须创建一个受限的执行环境。import ast import types class MockSandbox: 安全沙箱用于执行用户定义的AST规则 def __init__(self): # 1. 创建安全的全局命名空间 self._global_ns { ‘__builtins__‘: { # 限制内置函数只开放安全的 ‘len‘: len, ‘str‘: str, ‘int‘: int, ‘float‘: float, ‘bool‘: bool, ‘list‘: list, ‘dict‘: dict, ‘range‘: range, ‘sum‘: sum, ‘min‘: min, ‘max‘: max, ‘abs‘: abs, ‘round‘: round, }, ‘random‘: __import__(‘random‘), # 允许使用random模块生成随机数据 ‘datetime‘: __import__(‘datetime‘), # 允许生成时间数据 ‘json‘: __import__(‘json‘), ‘re‘: __import__(‘re‘), } # 明确禁止的模块或函数 self._forbidden {‘os‘, ‘sys‘, ‘subprocess‘, ‘__import__‘, ‘eval‘, ‘exec‘, ‘open‘, ‘compile‘} def safe_eval(self, expr_ast, local_ns): 在沙箱中执行一个AST表达式节点并返回结果 # 2. AST安全检查可选但推荐遍历AST检查是否有调用禁止的函数或属性 self._validate_ast(expr_ast) # 3. 编译AST为字节码 try: code_obj compile(expr_ast, ‘mock_rule‘, ‘eval‘) except SyntaxError as e: raise ValueError(f“规则语法错误: {e}”) # 4. 在沙箱全局空间和本次请求的局部空间内执行 try: result eval(code_obj, self._global_ns, local_ns) except Exception as e: # 捕获执行过程中的异常并转换为友好的Mock错误 raise RuntimeError(f“规则执行失败: {e}”) return result def _validate_ast(self, node): 递归遍历AST进行安全检查 if isinstance(node, ast.Call): # 检查函数调用 if isinstance(node.func, ast.Name): if node.func.id in self._forbidden: raise SecurityError(f“禁止调用函数: {node.func.id}”) elif isinstance(node.func, ast.Attribute): # 检查类似 os.system 的调用 module_name self._get_full_attribute_name(node.func) if any(forbidden in module_name for forbidden in self._forbidden): raise SecurityError(f“禁止访问属性: {module_name}”) # 递归检查所有子节点 for child in ast.iter_child_nodes(node): self._validate_ast(child) def _get_full_attribute_name(self, node): 获取属性访问的全名如 os.path.join if isinstance(node, ast.Name): return node.id elif isinstance(node, ast.Attribute): return f“{self._get_full_attribute_name(node.value)}.{node.attr}” return “”注意这里的沙箱并非绝对安全Python动态特性使得构建完美沙箱极其困难但足以防御绝大多数无意或初级的危险操作。对于生产环境如果规则完全来自不可信用户需要更严格的限制甚至考虑使用PyPy的沙箱或容器隔离。3.2 定义Mock规则与AST编译我们需要一个数据结构来承载用户对某个接口的Mock定义并将其中的条件、响应字符串编译成可执行的AST。import yaml import hashlib class MockRule: 代表一个接口的Mock规则 def __init__(self, config: dict): self.path config[‘path‘] # 接口路径支持正则 self.method config.get(‘method‘, ‘GET‘).upper() self._request_check_raw config.get(‘request_check‘) # 原始的校验字符串 self._response_raw config.get(‘response‘) # 原始的响应模板字符串 self.delay config.get(‘delay‘, 0) # 模拟延迟单位秒 self.status_code config.get(‘status_code‘, 200) self.headers config.get(‘headers‘, {}) # 编译后的AST和字节码缓存提升性能 self._request_check_ast None self._response_ast None self._compiled_request_check None self._compiled_response None self._compile_rules() def _compile_rules(self): 将字符串规则编译为AST sandbox MockSandbox() # 编译请求校验规则 if self._request_check_raw: try: # 解析为AST表达式 self._request_check_ast ast.parse(self._request_check_raw, mode‘eval‘) # 可选的进行一些AST转换或优化 # 例如将 request.json[‘a‘] 转换为更安全的访问形式 self._request_check_ast self._rewrite_request_access(self._request_check_ast) # 编译为字节码对象并缓存 self._compiled_request_check compile(self._request_check_ast, ‘request_check‘, ‘eval‘) except SyntaxError as e: raise ValueError(f“请求校验规则语法错误: {e}”) # 编译响应生成规则 if isinstance(self._response_raw, str): try: # 响应可能是一个表达式如字典也可能是一段代码需赋值 # 我们这里假设是表达式。如果是多行代码需要mode‘exec‘ self._response_ast ast.parse(self._response_raw, mode‘eval‘) self._response_ast self._rewrite_request_access(self._response_ast) self._compiled_response compile(self._response_ast, ‘response_gen‘, ‘eval‘) except SyntaxError: # 如果不是表达式尝试作为字面量如纯JSON字符串 try: # 尝试直接解析为Python对象对于简单JSON self._response_data yaml.safe_load(self._response_raw) self._compiled_response None # 标记为静态数据 except: raise ValueError(f“无法解析响应规则: {self._response_raw}”) else: # 响应直接是Python dict/list等 self._response_data self._response_raw self._compiled_response None def _rewrite_request_access(self, tree): 一个简单的AST转换器示例将 request.json[‘key‘] 重写为安全访问 class RequestAccessRewriter(ast.NodeTransformer): def visit_Subscript(self, node): # 检查是否是 request.json[...] 的形式 if (isinstance(node.value, ast.Attribute) and isinstance(node.value.value, ast.Name) and node.value.value.id ‘request‘ and node.value.attr ‘json‘): # 将其转换为一个函数调用例如 _safe_get(request.json, ‘key‘) # 这里我们简化处理直接返回原节点 # 更复杂的版本可以在这里处理默认值或异常 pass return node return RequestAccessRewriter().visit(tree) def matches(self, incoming_path, incoming_method): 检查传入的请求是否匹配此规则 # 这里可以实现简单的字符串匹配或正则匹配 # 示例简单前缀匹配实际应用可能需要更复杂的路由匹配 return incoming_method self.method and incoming_path.startswith(self.path) def execute_check(self, request_obj, sandbox: MockSandbox): 执行请求校验返回(bool, error_msg) if not self._compiled_request_check: return True, “” # 无校验规则直接通过 local_ns {‘request‘: request_obj} try: check_result sandbox.safe_eval(self._request_check_ast, local_ns) if not isinstance(check_result, bool): return False, f“校验规则必须返回布尔值但返回了 {type(check_result).__name__}” return check_result, “” except Exception as e: return False, f“校验执行异常: {e}” def generate_response(self, request_obj, sandbox: MockSandbox): 生成响应数据 if self._compiled_response is None: # 静态数据 return self._response_data local_ns {‘request‘: request_obj} try: response_data sandbox.safe_eval(self._response_ast, local_ns) return response_data except Exception as e: # 生成响应失败返回一个错误兜底 return {“error”: f“动态生成响应失败: {e}”}这个MockRule类完成了从配置到可执行单元的转换。它缓存了编译后的字节码避免了每次请求都重复进行parse和compile这对性能至关重要。3.3 构建规则匹配与执行引擎现在我们需要一个引擎来管理所有规则并在HTTP请求到来时找到匹配的规则并执行它。class MockEngine: Mock规则引擎负责管理所有规则并执行匹配 def __init__(self, rule_filesNone): self.sandbox MockSandbox() self.rules [] # 存储所有加载的MockRule对象 if rule_files: self.load_rules_from_files(rule_files) def load_rules_from_files(self, file_paths): 从YAML/JSON文件加载规则 for fp in file_paths: with open(fp, ‘r‘, encoding‘utf-8‘) as f: if fp.endswith(‘.yaml‘) or fp.endswith(‘.yml‘): configs yaml.safe_load(f) elif fp.endswith(‘.json‘): import json configs json.load(f) else: continue # 支持文件内定义多个规则 if isinstance(configs, list): for config in configs: self.add_rule(config) elif isinstance(configs, dict): self.add_rule(configs) def add_rule(self, config: dict): 添加一条规则 rule MockRule(config) self.rules.append(rule) # 可以按路径或优先级排序优化匹配速度 self.rules.sort(keylambda x: len(x.path), reverseTrue) # 简单按路径长度倒序实现类似最长前缀匹配 def find_and_execute(self, request_path, request_method, request_obj): 查找并执行匹配的规则 for rule in self.rules: if rule.matches(request_path, request_method): # 1. 执行请求校验 check_ok, check_msg rule.execute_check(request_obj, self.sandbox) if not check_ok: # 校验失败返回错误响应 return { “status_code“: 400, # 或自定义错误码 “headers“: {“Content-Type“: “application/json“}, “body“: {“error“: “请求校验失败“, “detail“: check_msg} } # 2. 模拟延迟 if rule.delay 0: import time time.sleep(rule.delay) # 3. 生成响应数据 response_data rule.generate_response(request_obj, self.sandbox) # 4. 包装响应 return { “status_code“: rule.status_code, “headers“: {**{“Content-Type“: “application/json“}, **rule.headers}, “body“: response_data } # 没有找到匹配的规则 return { “status_code“: 404, “headers“: {“Content-Type“: “application/json“}, “body“: {“error“: “未找到匹配的Mock规则“} }引擎的工作流程非常清晰匹配 - 校验 - 延迟 - 生成响应。它封装了所有复杂性对外提供一个简单的find_and_execute接口。4. 集成HTTP服务与完整示例核心引擎已经就绪现在我们需要给它套上一个HTTP服务器的“外壳”让它能真正处理网络请求。这里我们选择轻量级的aiohttp来构建一个异步服务器性能更好。4.1 使用aiohttp构建Mock服务器from aiohttp import web import json class MockServer: 基于aiohttp的Mock HTTP服务器 def __init__(self, engine: MockEngine, host‘127.0.0.1‘, port9999): self.engine engine self.host host self.port port self.app web.Application() self._setup_routes() def _setup_routes(self): # 添加一个中间件用于记录日志或处理异常 self.app.middlewares.append(self._error_middleware) # 核心将所有HTTP方法的路由都指向同一个处理函数 self.app.router.add_route(‘*‘, ‘/{path:.*}‘, self._handle_request) web.middleware async def _error_middleware(self, request, handler): try: return await handler(request) except Exception as e: return web.Response( status500, textjson.dumps({“error“: “服务器内部错误“, “detail“: str(e)}), content_type“application/json“ ) async def _handle_request(self, request): # 1. 收集请求信息构建给引擎的request_obj request_obj { “method“: request.method, “path“: request.path, “headers“: dict(request.headers), “query“: dict(request.query), # GET参数 “json“: None, “text“: None, “raw“: request # 保留原始request对象以备高级使用 } # 2. 尝试解析请求体 content_type request.headers.get(‘Content-Type‘, ‘‘).lower() if ‘application/json‘ in content_type: try: request_obj[‘json‘] await request.json() except: pass # 解析失败则保持为None elif ‘text/‘ in content_type or ‘application/x-www-form-urlencoded‘ in content_type: request_obj[‘text‘] await request.text() # 3. 调用引擎处理 result self.engine.find_and_execute( request_pathrequest.path, request_methodrequest.method, request_objrequest_obj ) # 4. 返回HTTP响应 body result.get(‘body‘, {}) if isinstance(body, (dict, list)): body json.dumps(body, ensure_asciiFalse) return web.Response( statusresult.get(‘status_code‘, 200), textbody, headersresult.get(‘headers‘, {}) ) def run(self): 启动服务器 web.run_app(self.app, hostself.host, portself.port)4.2 编写Mock规则定义文件让我们用一个完整的YAML示例来展示这个Mock工具的强大之处。# mock_rules.yaml - path: /api/v1/login method: POST request_check: | # 校验请求体为JSON且包含必要字段 isinstance(request.json, dict) and ‘username‘ in request.json and ‘password‘ in request.json and len(request.json[‘username‘]) 3 response: | { “success“: request.json[‘password‘] “123456“, # 动态判断密码 “user_id“: random.randint(1000, 9999) if request.json[‘password‘] “123456“ else None, “token“: “mock_token_“ str(random.randint(10000, 99999)) if request.json[‘password‘] “123456“ else None, “message“: “登录成功“ if request.json[‘password‘] “123456“ else “密码错误“ } delay: 0.5 # 模拟0.5秒网络延迟 status_code: 200 - path: /api/v1/orders method: GET request_check: “ ‘user_id‘ in request.query and request.query[‘user_id‘].isdigit() “ response: | { “orders“: [ { “id“: i, “amount“: random.randint(100, 5000), “status“: random.choice([“pending“, “shipped“, “delivered“]), “created_at“: (datetime.datetime.now() - datetime.timedelta(daysi)).isoformat() } for i in range(1, 4) # 根据user_id动态生成3个订单这里简化了 ] } - path: /api/v1/upload method: POST request_check: “ ‘file‘ in request.raw“ # 这里演示访问原始request对象需要额外处理 response: | { “url“: f“https://mock-cdn.com/{hashlib.md5(str(random.random()).encode()).hexdigest()}.jpg“ } headers: X-Custom-Header: Mocked4.3 启动与测试最后写一个主程序把一切串起来。# main.py import asyncio from your_mock_module import MockEngine, MockServer # 替换为你的模块名 def main(): # 1. 初始化引擎并加载规则 engine MockEngine() engine.load_rules_from_files([‘mock_rules.yaml‘]) # 2. 可以动态添加规则 engine.add_rule({ “path“: “/api/health“, “method“: “GET“, “response“: {“status“: “ok“, “service“: “mock-server“} }) # 3. 创建并启动服务器 server MockServer(engine, host‘0.0.0.0‘, port8080) print(f“Mock服务器启动在 http://{server.host}:{server.port}“) print(“已加载规则:“) for rule in engine.rules: print(f“ {rule.method} {rule.path}“) server.run() if __name__ ‘__main__‘: main()启动后你就可以用curl、Postman或任何HTTP客户端进行测试了。# 测试登录接口 curl -X POST http://127.0.0.1:8080/api/v1/login \ -H “Content-Type: application/json“ \ -d ‘{“username“: “test“, “password“: “123456“}‘ # 预期返回 # { # “success“: true, # “user_id“: 7421, # “token“: “mock_token_88462“, # “message“: “登录成功“ # } # 测试错误密码 curl -X POST http://127.0.0.1:8080/api/v1/login \ -H “Content-Type: application/json“ \ -d ‘{“username“: “test“, “password“: “wrong“}‘ # 测试订单查询 curl “http://127.0.0.1:8080/api/v1/orders?user_id100“5. 高级特性、问题排查与优化建议一个基础可用的Mock工具已经完成了。但在实际生产或复杂场景中使用我们还需要考虑更多。5.1 实现更强大的规则匹配与优先级目前的matches方法只是简单的前缀匹配。在实际项目中你可能需要正则表达式匹配path字段可以是一个正则模式字符串。路径参数提取例如将/api/users/{user_id}/orders中的{user_id}提取出来放入request_obj供规则使用。规则优先级当多个规则匹配同一个请求时比如一个通用规则/api/*和一个具体规则/api/user需要定义优先级。可以在MockRule中增加一个priority字段并在引擎匹配时优先选择优先级高或路径更具体的规则。# 在MockRule.__init__中 self.priority config.get(‘priority‘, 0) # 数值越大优先级越高 # 在MockEngine.find_and_execute中匹配后不立即返回而是收集所有匹配规则 matched_rules [rule for rule in self.rules if rule.matches(...)] if matched_rules: # 按优先级和路径特异性排序 matched_rules.sort(keylambda x: (x.priority, -len(x.path)), reverseTrue) rule matched_rules[0] # 选择最高优先级的规则 # ... 执行该规则5.2 请求/响应数据的转换与验证JSON Schema校验对于复杂的请求体校验单纯靠Python表达式会很长。可以集成jsonschema库在request_check中支持schema字段。request_check: schema: type: object required: [“name“, “email“] properties: name: {type: “string“} email: {type: “string“, format: “email“}引擎需要识别这种格式并调用jsonschema.validate。响应模板引擎除了Python表达式可以支持简单的模板语法如{request.json[‘name’]}这需要在generate_response中增加一个模板渲染步骤。5.3 常见问题排查与调试技巧规则不生效检查路径和方法首先用print确认请求的path和method是否完全匹配规则定义。注意URL末尾的/。检查规则加载顺序确保规则文件被正确加载没有YAML/JSON语法错误。在MockEngine初始化后打印self.rules的长度和内容。启用调试日志在MockServer._handle_request和MockEngine.find_and_execute中加入日志打印匹配过程和执行结果。AST编译或执行错误语法错误用户写的规则字符串可能有Python语法错误。确保用ast.parse进行解析并捕获SyntaxError给出友好提示。沙箱安全错误用户规则尝试调用了禁止的函数。错误信息应明确指出是哪一行代码、哪个函数被禁止了。运行时错误规则逻辑本身有错比如访问了request.json但请求体不是JSON。在safe_eval中要用try-except包裹并返回详细的错误信息给用户。性能问题AST编译缓存我们已经做了。确保MockRule._compile_rules只在规则初始化时调用一次。规则匹配优化如果规则很多上百条线性遍历self.rules会成为瓶颈。可以考虑使用字典按(method, path_prefix)建立索引或者使用trie树前缀树来加速路径匹配。响应延迟time.sleep会阻塞整个线程在异步服务器中尤其糟糕。对于异步服务器如aiohttp务必使用asyncio.sleep。5.4 扩展方向让它更“企业级”规则的热重载监听规则文件变化无需重启服务即可更新Mock行为。可以使用watchdog库。请求录制与回放将真实的线上流量录制下来自动或半自动地生成Mock规则。这是构建“契约测试”或“服务虚拟化”的基础。集成到CI/CD将Mock服务器作为测试套件的一部分启动用于接口自动化测试。提供Web管理界面使用Flask或FastAPI快速搭建一个界面方便非开发人员如测试、产品查看、编辑和启停Mock规则。支持更多协议除了HTTP/HTTPS还可以扩展支持gRPC、WebSocket等协议的Mock。从头构建一个基于AST的Mock工具这个过程本身就是一个绝佳的学习之旅。你不仅深入理解了Mock的各个层面还亲手实践了AST编译、沙箱安全、Web服务器等多项核心技能。这个工具可能一开始只是为了解决你手头的联调问题但随着不断打磨和扩展它有潜力成长为你团队甚至公司内部的一个强大基础设施。最重要的是你掌握了“创造工具”的思维和能力这才是开发者最宝贵的财富。 基于AST构建Python接口Mock工具：从原理到工程实践
1. 项目概述为什么我们需要自己造一个Mock工具在前后端分离、微服务架构大行其道的今天接口联调是每个开发者都绕不开的坎。你肯定遇到过这种情况前端页面急着要上线后端接口还在开发中或者依赖的第三方服务不稳定动不动就给你来个超时。这时候一个靠谱的接口Mock工具就成了救命稻草。市面上的Mock工具不少像Postman的Mock Server、Apifox、YApi还有Python里的responses、httpretty这些库用起来都挺方便。那为什么还要自己动手写一个呢原因很简单不够“趁手”。现有的工具要么太重配置繁琐要么太轻功能单一比如只能返回固定数据无法根据请求参数动态响应更别提对请求体做复杂的校验了。特别是当你需要模拟一个逻辑稍微复杂一点的接口比如请求参数A大于10时返回成功否则返回失败或者需要校验JSON请求体里某个嵌套字段的格式时通用工具往往需要你写一大段脚本或者干脆做不到。这就是“手写Python接口Mock工具”的价值所在。我们不是要做一个替代所有现有工具的巨无霸而是要打造一把高度定制化、深度可控的“瑞士军刀”。基于AST抽象语法树来实现听起来有点“硬核”但这恰恰是它的精髓。AST允许我们以代码结构本身为操作对象这意味着我们可以用写Python逻辑的方式去定义Mock的规则——校验请求、生成响应、处理异常所有这些都能用你熟悉的if-else、循环、函数调用来表达灵活性和可编程性直接拉满。这个项目就是带你从零开始理解Mock的核心需求并用AST这把“手术刀”精细地构建一个属于你自己的、功能强大的Mock工具。2. 核心需求与设计思路拆解在动手写代码之前我们必须把需求掰开揉碎了看。一个功能完备的Mock工具远不止是返回一个{code: 200, “data”: “hello”}这么简单。2.1 核心需求解析接口模拟这是基础。工具必须能监听特定端口和路径对HTTP请求GET、POST等做出响应。它需要模拟真实接口的延迟、状态码、响应头和响应体。请求校验这是保证Mock有效性的关键。我们需要能验证请求的方方面面结构校验请求体是否是合法的JSONXML字段校验某个必填字段是否存在字段的数据类型字符串、数字、数组是否正确业务逻辑校验字段A的值是否在某个范围内字段B是否依赖于字段A的值例如创建订单时金额必须大于0。动态响应这是Mock工具智能化的体现。响应不能是死的它应该能根据请求的内容动态变化。参数化响应将请求中的某个值如user_id填充到响应模板中。条件化响应根据不同的请求参数或头部信息返回完全不同的响应内容和状态码。逻辑生成响应数据可以通过一段逻辑代码计算得出比如生成一个随机的订单号或者计算一个基于请求参数的哈希值。2.2 为什么选择基于AST的实现方案面对这些需求常见的实现思路可能是用一堆if语句和正则表达式在请求处理函数里硬编码。但这样代码会迅速变得臃肿且难以维护特别是当Mock规则复杂时。AST方案的优势在于“声明式”与“运行时编译”。声明式配置我们可以让用户用一种接近自然语言或简化Python语法的方式比如一个字典或一段特定格式的字符串来定义Mock规则。例如{ “path”: “/api/user”, “method”: “POST”, “request_check”: “request.json[‘age’] 0 and len(request.json[‘name’]) 1”, “response”: “{‘id’: random.randint(1000, 9999), ‘name’: request.json[‘name’]}” }AST转换与执行上面的request_check和response是字符串。我们的工具核心就是将这些字符串解析成AST然后在处理真实请求的上下文中编译并执行这棵语法树。request.json[‘age’]在AST中就是一个属性访问节点是一个比较操作节点。执行时request会被替换为实际的请求对象。这样做的好处是安全通过限制AST可访问的变量和函数__builtins__我们可以构建一个沙箱环境防止用户定义的Mock规则执行危险代码如os.system(‘rm -rf /’)。灵活理论上用户可以用任何合法的Python表达式来定义规则能力边界非常广。性能AST可以被编译成字节码code object并缓存。对于高频调用的Mock接口首次编译后后续执行就是直接的字节码解释速度很快。设计蓝图我们的工具将分为三层。配置层负责读取和解析用户定义的规则文件YAML/JSON。核心引擎层包含AST解析器、安全沙箱构造器、规则编译器和执行器。这是最核心的部分。HTTP服务层一个轻量的Web服务器如使用aiohttp或Flask接收请求将请求对象传递给核心引擎匹配并执行规则最后返回引擎生成的响应。3. 核心引擎实现AST的解析、沙箱与执行这是整个项目最硬核、也最有趣的部分。我们将一步步构建核心引擎。3.1 构建安全的AST执行沙箱直接执行来自用户的字符串代码是极度危险的。我们必须创建一个受限的执行环境。import ast import types class MockSandbox: 安全沙箱用于执行用户定义的AST规则 def __init__(self): # 1. 创建安全的全局命名空间 self._global_ns { ‘__builtins__‘: { # 限制内置函数只开放安全的 ‘len‘: len, ‘str‘: str, ‘int‘: int, ‘float‘: float, ‘bool‘: bool, ‘list‘: list, ‘dict‘: dict, ‘range‘: range, ‘sum‘: sum, ‘min‘: min, ‘max‘: max, ‘abs‘: abs, ‘round‘: round, }, ‘random‘: __import__(‘random‘), # 允许使用random模块生成随机数据 ‘datetime‘: __import__(‘datetime‘), # 允许生成时间数据 ‘json‘: __import__(‘json‘), ‘re‘: __import__(‘re‘), } # 明确禁止的模块或函数 self._forbidden {‘os‘, ‘sys‘, ‘subprocess‘, ‘__import__‘, ‘eval‘, ‘exec‘, ‘open‘, ‘compile‘} def safe_eval(self, expr_ast, local_ns): 在沙箱中执行一个AST表达式节点并返回结果 # 2. AST安全检查可选但推荐遍历AST检查是否有调用禁止的函数或属性 self._validate_ast(expr_ast) # 3. 编译AST为字节码 try: code_obj compile(expr_ast, ‘mock_rule‘, ‘eval‘) except SyntaxError as e: raise ValueError(f“规则语法错误: {e}”) # 4. 在沙箱全局空间和本次请求的局部空间内执行 try: result eval(code_obj, self._global_ns, local_ns) except Exception as e: # 捕获执行过程中的异常并转换为友好的Mock错误 raise RuntimeError(f“规则执行失败: {e}”) return result def _validate_ast(self, node): 递归遍历AST进行安全检查 if isinstance(node, ast.Call): # 检查函数调用 if isinstance(node.func, ast.Name): if node.func.id in self._forbidden: raise SecurityError(f“禁止调用函数: {node.func.id}”) elif isinstance(node.func, ast.Attribute): # 检查类似 os.system 的调用 module_name self._get_full_attribute_name(node.func) if any(forbidden in module_name for forbidden in self._forbidden): raise SecurityError(f“禁止访问属性: {module_name}”) # 递归检查所有子节点 for child in ast.iter_child_nodes(node): self._validate_ast(child) def _get_full_attribute_name(self, node): 获取属性访问的全名如 os.path.join if isinstance(node, ast.Name): return node.id elif isinstance(node, ast.Attribute): return f“{self._get_full_attribute_name(node.value)}.{node.attr}” return “”注意这里的沙箱并非绝对安全Python动态特性使得构建完美沙箱极其困难但足以防御绝大多数无意或初级的危险操作。对于生产环境如果规则完全来自不可信用户需要更严格的限制甚至考虑使用PyPy的沙箱或容器隔离。3.2 定义Mock规则与AST编译我们需要一个数据结构来承载用户对某个接口的Mock定义并将其中的条件、响应字符串编译成可执行的AST。import yaml import hashlib class MockRule: 代表一个接口的Mock规则 def __init__(self, config: dict): self.path config[‘path‘] # 接口路径支持正则 self.method config.get(‘method‘, ‘GET‘).upper() self._request_check_raw config.get(‘request_check‘) # 原始的校验字符串 self._response_raw config.get(‘response‘) # 原始的响应模板字符串 self.delay config.get(‘delay‘, 0) # 模拟延迟单位秒 self.status_code config.get(‘status_code‘, 200) self.headers config.get(‘headers‘, {}) # 编译后的AST和字节码缓存提升性能 self._request_check_ast None self._response_ast None self._compiled_request_check None self._compiled_response None self._compile_rules() def _compile_rules(self): 将字符串规则编译为AST sandbox MockSandbox() # 编译请求校验规则 if self._request_check_raw: try: # 解析为AST表达式 self._request_check_ast ast.parse(self._request_check_raw, mode‘eval‘) # 可选的进行一些AST转换或优化 # 例如将 request.json[‘a‘] 转换为更安全的访问形式 self._request_check_ast self._rewrite_request_access(self._request_check_ast) # 编译为字节码对象并缓存 self._compiled_request_check compile(self._request_check_ast, ‘request_check‘, ‘eval‘) except SyntaxError as e: raise ValueError(f“请求校验规则语法错误: {e}”) # 编译响应生成规则 if isinstance(self._response_raw, str): try: # 响应可能是一个表达式如字典也可能是一段代码需赋值 # 我们这里假设是表达式。如果是多行代码需要mode‘exec‘ self._response_ast ast.parse(self._response_raw, mode‘eval‘) self._response_ast self._rewrite_request_access(self._response_ast) self._compiled_response compile(self._response_ast, ‘response_gen‘, ‘eval‘) except SyntaxError: # 如果不是表达式尝试作为字面量如纯JSON字符串 try: # 尝试直接解析为Python对象对于简单JSON self._response_data yaml.safe_load(self._response_raw) self._compiled_response None # 标记为静态数据 except: raise ValueError(f“无法解析响应规则: {self._response_raw}”) else: # 响应直接是Python dict/list等 self._response_data self._response_raw self._compiled_response None def _rewrite_request_access(self, tree): 一个简单的AST转换器示例将 request.json[‘key‘] 重写为安全访问 class RequestAccessRewriter(ast.NodeTransformer): def visit_Subscript(self, node): # 检查是否是 request.json[...] 的形式 if (isinstance(node.value, ast.Attribute) and isinstance(node.value.value, ast.Name) and node.value.value.id ‘request‘ and node.value.attr ‘json‘): # 将其转换为一个函数调用例如 _safe_get(request.json, ‘key‘) # 这里我们简化处理直接返回原节点 # 更复杂的版本可以在这里处理默认值或异常 pass return node return RequestAccessRewriter().visit(tree) def matches(self, incoming_path, incoming_method): 检查传入的请求是否匹配此规则 # 这里可以实现简单的字符串匹配或正则匹配 # 示例简单前缀匹配实际应用可能需要更复杂的路由匹配 return incoming_method self.method and incoming_path.startswith(self.path) def execute_check(self, request_obj, sandbox: MockSandbox): 执行请求校验返回(bool, error_msg) if not self._compiled_request_check: return True, “” # 无校验规则直接通过 local_ns {‘request‘: request_obj} try: check_result sandbox.safe_eval(self._request_check_ast, local_ns) if not isinstance(check_result, bool): return False, f“校验规则必须返回布尔值但返回了 {type(check_result).__name__}” return check_result, “” except Exception as e: return False, f“校验执行异常: {e}” def generate_response(self, request_obj, sandbox: MockSandbox): 生成响应数据 if self._compiled_response is None: # 静态数据 return self._response_data local_ns {‘request‘: request_obj} try: response_data sandbox.safe_eval(self._response_ast, local_ns) return response_data except Exception as e: # 生成响应失败返回一个错误兜底 return {“error”: f“动态生成响应失败: {e}”}这个MockRule类完成了从配置到可执行单元的转换。它缓存了编译后的字节码避免了每次请求都重复进行parse和compile这对性能至关重要。3.3 构建规则匹配与执行引擎现在我们需要一个引擎来管理所有规则并在HTTP请求到来时找到匹配的规则并执行它。class MockEngine: Mock规则引擎负责管理所有规则并执行匹配 def __init__(self, rule_filesNone): self.sandbox MockSandbox() self.rules [] # 存储所有加载的MockRule对象 if rule_files: self.load_rules_from_files(rule_files) def load_rules_from_files(self, file_paths): 从YAML/JSON文件加载规则 for fp in file_paths: with open(fp, ‘r‘, encoding‘utf-8‘) as f: if fp.endswith(‘.yaml‘) or fp.endswith(‘.yml‘): configs yaml.safe_load(f) elif fp.endswith(‘.json‘): import json configs json.load(f) else: continue # 支持文件内定义多个规则 if isinstance(configs, list): for config in configs: self.add_rule(config) elif isinstance(configs, dict): self.add_rule(configs) def add_rule(self, config: dict): 添加一条规则 rule MockRule(config) self.rules.append(rule) # 可以按路径或优先级排序优化匹配速度 self.rules.sort(keylambda x: len(x.path), reverseTrue) # 简单按路径长度倒序实现类似最长前缀匹配 def find_and_execute(self, request_path, request_method, request_obj): 查找并执行匹配的规则 for rule in self.rules: if rule.matches(request_path, request_method): # 1. 执行请求校验 check_ok, check_msg rule.execute_check(request_obj, self.sandbox) if not check_ok: # 校验失败返回错误响应 return { “status_code“: 400, # 或自定义错误码 “headers“: {“Content-Type“: “application/json“}, “body“: {“error“: “请求校验失败“, “detail“: check_msg} } # 2. 模拟延迟 if rule.delay 0: import time time.sleep(rule.delay) # 3. 生成响应数据 response_data rule.generate_response(request_obj, self.sandbox) # 4. 包装响应 return { “status_code“: rule.status_code, “headers“: {**{“Content-Type“: “application/json“}, **rule.headers}, “body“: response_data } # 没有找到匹配的规则 return { “status_code“: 404, “headers“: {“Content-Type“: “application/json“}, “body“: {“error“: “未找到匹配的Mock规则“} }引擎的工作流程非常清晰匹配 - 校验 - 延迟 - 生成响应。它封装了所有复杂性对外提供一个简单的find_and_execute接口。4. 集成HTTP服务与完整示例核心引擎已经就绪现在我们需要给它套上一个HTTP服务器的“外壳”让它能真正处理网络请求。这里我们选择轻量级的aiohttp来构建一个异步服务器性能更好。4.1 使用aiohttp构建Mock服务器from aiohttp import web import json class MockServer: 基于aiohttp的Mock HTTP服务器 def __init__(self, engine: MockEngine, host‘127.0.0.1‘, port9999): self.engine engine self.host host self.port port self.app web.Application() self._setup_routes() def _setup_routes(self): # 添加一个中间件用于记录日志或处理异常 self.app.middlewares.append(self._error_middleware) # 核心将所有HTTP方法的路由都指向同一个处理函数 self.app.router.add_route(‘*‘, ‘/{path:.*}‘, self._handle_request) web.middleware async def _error_middleware(self, request, handler): try: return await handler(request) except Exception as e: return web.Response( status500, textjson.dumps({“error“: “服务器内部错误“, “detail“: str(e)}), content_type“application/json“ ) async def _handle_request(self, request): # 1. 收集请求信息构建给引擎的request_obj request_obj { “method“: request.method, “path“: request.path, “headers“: dict(request.headers), “query“: dict(request.query), # GET参数 “json“: None, “text“: None, “raw“: request # 保留原始request对象以备高级使用 } # 2. 尝试解析请求体 content_type request.headers.get(‘Content-Type‘, ‘‘).lower() if ‘application/json‘ in content_type: try: request_obj[‘json‘] await request.json() except: pass # 解析失败则保持为None elif ‘text/‘ in content_type or ‘application/x-www-form-urlencoded‘ in content_type: request_obj[‘text‘] await request.text() # 3. 调用引擎处理 result self.engine.find_and_execute( request_pathrequest.path, request_methodrequest.method, request_objrequest_obj ) # 4. 返回HTTP响应 body result.get(‘body‘, {}) if isinstance(body, (dict, list)): body json.dumps(body, ensure_asciiFalse) return web.Response( statusresult.get(‘status_code‘, 200), textbody, headersresult.get(‘headers‘, {}) ) def run(self): 启动服务器 web.run_app(self.app, hostself.host, portself.port)4.2 编写Mock规则定义文件让我们用一个完整的YAML示例来展示这个Mock工具的强大之处。# mock_rules.yaml - path: /api/v1/login method: POST request_check: | # 校验请求体为JSON且包含必要字段 isinstance(request.json, dict) and ‘username‘ in request.json and ‘password‘ in request.json and len(request.json[‘username‘]) 3 response: | { “success“: request.json[‘password‘] “123456“, # 动态判断密码 “user_id“: random.randint(1000, 9999) if request.json[‘password‘] “123456“ else None, “token“: “mock_token_“ str(random.randint(10000, 99999)) if request.json[‘password‘] “123456“ else None, “message“: “登录成功“ if request.json[‘password‘] “123456“ else “密码错误“ } delay: 0.5 # 模拟0.5秒网络延迟 status_code: 200 - path: /api/v1/orders method: GET request_check: “ ‘user_id‘ in request.query and request.query[‘user_id‘].isdigit() “ response: | { “orders“: [ { “id“: i, “amount“: random.randint(100, 5000), “status“: random.choice([“pending“, “shipped“, “delivered“]), “created_at“: (datetime.datetime.now() - datetime.timedelta(daysi)).isoformat() } for i in range(1, 4) # 根据user_id动态生成3个订单这里简化了 ] } - path: /api/v1/upload method: POST request_check: “ ‘file‘ in request.raw“ # 这里演示访问原始request对象需要额外处理 response: | { “url“: f“https://mock-cdn.com/{hashlib.md5(str(random.random()).encode()).hexdigest()}.jpg“ } headers: X-Custom-Header: Mocked4.3 启动与测试最后写一个主程序把一切串起来。# main.py import asyncio from your_mock_module import MockEngine, MockServer # 替换为你的模块名 def main(): # 1. 初始化引擎并加载规则 engine MockEngine() engine.load_rules_from_files([‘mock_rules.yaml‘]) # 2. 可以动态添加规则 engine.add_rule({ “path“: “/api/health“, “method“: “GET“, “response“: {“status“: “ok“, “service“: “mock-server“} }) # 3. 创建并启动服务器 server MockServer(engine, host‘0.0.0.0‘, port8080) print(f“Mock服务器启动在 http://{server.host}:{server.port}“) print(“已加载规则:“) for rule in engine.rules: print(f“ {rule.method} {rule.path}“) server.run() if __name__ ‘__main__‘: main()启动后你就可以用curl、Postman或任何HTTP客户端进行测试了。# 测试登录接口 curl -X POST http://127.0.0.1:8080/api/v1/login \ -H “Content-Type: application/json“ \ -d ‘{“username“: “test“, “password“: “123456“}‘ # 预期返回 # { # “success“: true, # “user_id“: 7421, # “token“: “mock_token_88462“, # “message“: “登录成功“ # } # 测试错误密码 curl -X POST http://127.0.0.1:8080/api/v1/login \ -H “Content-Type: application/json“ \ -d ‘{“username“: “test“, “password“: “wrong“}‘ # 测试订单查询 curl “http://127.0.0.1:8080/api/v1/orders?user_id100“5. 高级特性、问题排查与优化建议一个基础可用的Mock工具已经完成了。但在实际生产或复杂场景中使用我们还需要考虑更多。5.1 实现更强大的规则匹配与优先级目前的matches方法只是简单的前缀匹配。在实际项目中你可能需要正则表达式匹配path字段可以是一个正则模式字符串。路径参数提取例如将/api/users/{user_id}/orders中的{user_id}提取出来放入request_obj供规则使用。规则优先级当多个规则匹配同一个请求时比如一个通用规则/api/*和一个具体规则/api/user需要定义优先级。可以在MockRule中增加一个priority字段并在引擎匹配时优先选择优先级高或路径更具体的规则。# 在MockRule.__init__中 self.priority config.get(‘priority‘, 0) # 数值越大优先级越高 # 在MockEngine.find_and_execute中匹配后不立即返回而是收集所有匹配规则 matched_rules [rule for rule in self.rules if rule.matches(...)] if matched_rules: # 按优先级和路径特异性排序 matched_rules.sort(keylambda x: (x.priority, -len(x.path)), reverseTrue) rule matched_rules[0] # 选择最高优先级的规则 # ... 执行该规则5.2 请求/响应数据的转换与验证JSON Schema校验对于复杂的请求体校验单纯靠Python表达式会很长。可以集成jsonschema库在request_check中支持schema字段。request_check: schema: type: object required: [“name“, “email“] properties: name: {type: “string“} email: {type: “string“, format: “email“}引擎需要识别这种格式并调用jsonschema.validate。响应模板引擎除了Python表达式可以支持简单的模板语法如{request.json[‘name’]}这需要在generate_response中增加一个模板渲染步骤。5.3 常见问题排查与调试技巧规则不生效检查路径和方法首先用print确认请求的path和method是否完全匹配规则定义。注意URL末尾的/。检查规则加载顺序确保规则文件被正确加载没有YAML/JSON语法错误。在MockEngine初始化后打印self.rules的长度和内容。启用调试日志在MockServer._handle_request和MockEngine.find_and_execute中加入日志打印匹配过程和执行结果。AST编译或执行错误语法错误用户写的规则字符串可能有Python语法错误。确保用ast.parse进行解析并捕获SyntaxError给出友好提示。沙箱安全错误用户规则尝试调用了禁止的函数。错误信息应明确指出是哪一行代码、哪个函数被禁止了。运行时错误规则逻辑本身有错比如访问了request.json但请求体不是JSON。在safe_eval中要用try-except包裹并返回详细的错误信息给用户。性能问题AST编译缓存我们已经做了。确保MockRule._compile_rules只在规则初始化时调用一次。规则匹配优化如果规则很多上百条线性遍历self.rules会成为瓶颈。可以考虑使用字典按(method, path_prefix)建立索引或者使用trie树前缀树来加速路径匹配。响应延迟time.sleep会阻塞整个线程在异步服务器中尤其糟糕。对于异步服务器如aiohttp务必使用asyncio.sleep。5.4 扩展方向让它更“企业级”规则的热重载监听规则文件变化无需重启服务即可更新Mock行为。可以使用watchdog库。请求录制与回放将真实的线上流量录制下来自动或半自动地生成Mock规则。这是构建“契约测试”或“服务虚拟化”的基础。集成到CI/CD将Mock服务器作为测试套件的一部分启动用于接口自动化测试。提供Web管理界面使用Flask或FastAPI快速搭建一个界面方便非开发人员如测试、产品查看、编辑和启停Mock规则。支持更多协议除了HTTP/HTTPS还可以扩展支持gRPC、WebSocket等协议的Mock。从头构建一个基于AST的Mock工具这个过程本身就是一个绝佳的学习之旅。你不仅深入理解了Mock的各个层面还亲手实践了AST编译、沙箱安全、Web服务器等多项核心技能。这个工具可能一开始只是为了解决你手头的联调问题但随着不断打磨和扩展它有潜力成长为你团队甚至公司内部的一个强大基础设施。最重要的是你掌握了“创造工具”的思维和能力这才是开发者最宝贵的财富。