手写静态Stub：固定响应实现、HTTP/Thrift/gRPC协议对接

原理概述

静态 Stub（Static Stub）是模块边界测试中最直接、最可控的桩实现方式。它不包含任何业务逻辑分支，对于每个被调用的接口方法，始终返回一个预定义的、固定的响应。这种“死数据”策略在以下场景中非常有效：

  • 测试需要完全消除下游不确定性。
  • 被测模块仅关心自身对请求的组装与发送，不处理响应变化。
  • 被测模块的异常逻辑（如超时回退）尚未实现，需要先验证正常路径。

与动态 Stub（根据输入不同返回不同数据）相比，静态 Stub 的代码量更少、维护成本极低，但灵活性受限。当被测模块需要验证多种响应结果（200、404、业务错误码）时，通常需要部署多个不同端口上的静态 Stub 实例，或通过环境变量切换固定响应。

实现流程

  1. 确定协议与接口定义：根据被测模块实际调用的下游接口，选择 HTTP REST、Thrift 或 gRPC 协议，并获取其 IDL（如 proto 文件、Thrift IDL、OpenAPI 规范）。
  2. 生成服务框架代码：使用对应工具（protoc、thrift 编译器、框架脚手架）生成服务端骨架。
  3. 编写固定响应逻辑：在每个 RPC 方法或路由处理函数中，直接返回一个硬编码的响应体（JSON 字符串、Thrift 结构体、protobuf message）。
  4. 配置参数与启动：设置监听地址、端口、可选的超时设置，启动 Stub 进程。
  5. 验证响应正确性：使用客户端工具（curl、grpcurl、Thrift 客户端）发送请求，确认返回数据与预期一致。
最佳实践：将固定响应内容外置于配置文件（JSON/YAML）中，通过启动参数或环境变量选择不同的响应集合，实现“多组固定响应”的切换，无需重新编译代码。

协议对接实现

以下是使用 Python 实现三种协议静态 Stub 的典型代码片段。为保持简洁，所有示例仅展示核心的固定响应逻辑，省略了完整框架代码。

HTTP REST 静态 Stub（基于 Flask）

from flask import Flask, jsonify, request

app = Flask(__name__)

# 配置固定响应（可从文件读取）
FIXED_PAYLOAD = {"status": "ok", "result": 42}

@app.route('/api/v1/process', methods=['POST'])
def process():
    # 忽略请求体，直接返回固定响应
    return jsonify(FIXED_PAYLOAD), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=9080)

Thrift 静态 Stub（Python + Thriftpy2）

# calculator.thrift
service Calculator {
    i32 add(1: i32 a, 2: i32 b)
}

# stub_server.py
import thriftpy2
from thriftpy2.thrift import TProcessor
from thriftpy2.transport import TSocket, TServerSocket, TTransport
from thriftpy2.protocol import TBinaryProtocolFactory

calc_thrift = thriftpy2.load("calculator.thrift", module_name="calc_thrift")

class FixedHandler:
    def add(self, a, b):
        # 固定返回 0，忽略输入
        return 0

handler = FixedHandler()
processor = TProcessor(calc_thrift.Calculator, handler)
transport = TServerSocket(host='0.0.0.0', port=9090)
tfactory = TTransport.TBufferedTransportFactory()
pfactory = TBinaryProtocolFactory()

server = TServer.TSimpleServer(processor, transport, tfactory, pfactory)
server.serve()

gRPC 静态 Stub（Python + grpcio）

# math.proto
service Math {
    rpc Multiply (MultiplyRequest) returns (MultiplyResponse);
}
message MultiplyRequest { int64 a = 1; int64 b = 2; }
message MultiplyResponse { int64 result = 1; }

# stub_server.py
import grpc
from concurrent import futures
import math_pb2
import math_pb2_grpc

class FixedMathServicer(math_pb2_grpc.MathServicer):
    def Multiply(self, request, context):
        # 固定返回 result = 100
        return math_pb2.MultiplyResponse(result=100)

server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
math_pb2_grpc.add_MathServicer_to_server(FixedMathServicer(), server)
server.add_insecure_port('[::]:50051')
server.start()
server.wait_for_termination()

内联 SVG：静态 Stub 协议适配流程

静态 Stub 协议适配流程 测试请求 （任意输入，Stub 忽略） HTTP REST Flask / FastAPI Thrift TProcessor gRPC Servicer 静态 Stub 核心 返回硬编码固定响应 固定响应配置表（JSON / YAML） 协议、方法、期望状态码、返回体

调试与验证

协议 验证命令 预期输出
HTTP 
curl -X POST http://localhost:9080/api/v1/process -d '{"a":1}'
 
{"status": "ok", "result": 42}
Thrift 
thrift-client --host localhost --port 9090 --protocol binary --service Calculator --method add --args 123,456
 
0
（固定返回值）
gRPC 
grpcurl -plaintext -d '{"a":5,"b":7}' localhost:50051 math.Math/Multiply
 
{"result": 100}

部署与运行

  • 独立进程：每个 Stub 可单独启动，绑定不同端口。
  • 容器化：构建最小 Docker 镜像，暴露对应端口，通过环境变量注入固定响应。
  • 集成测试套件：在 CI 中启动 Stub 后运行被测模块的集成测试，结束后关闭。
提示：gRPC Stub 需要确保 proto 文件版本与客户端完全一致，否则序列化失败。推荐将 proto 文件作为依赖统一管理。

常见错误与解决方法

错误现象原因解决
HTTP 返回 404 路由路径不匹配 检查 Flask 路由定义，确认方法（GET/POST）一致
Thrift TApplicationException 方法签名不匹配或版本不兼容 重新生成处理器代码，核对 thrift IDL
gRPC 无法反序列化 protobuf message 字段类型不符 检查 proto 定义与固定响应的 message 类型是否对应
端口冲突 多个 Stub 实例绑定同一端口 每次启动使用不同的端口号

性能与稳定性优化

  • 无状态设计：静态 Stub 不应保存任何请求上下文，避免内存泄漏。
  • 连接池与协程：对于高并发场景，gRPC 和 HTTP 可使用异步框架（aiohttp、grpc.aio）提升吞吐。
  • 日志采样：固定响应 Stub 的日志量可能较大，建议按比例采样或仅记录错误。
  • 健康检查端点：提供 /health 端点供编排工具检测 Stub 是否存活。
← 上一章 📖 返回目录 下一章 →