商城接口自动化：Mitmproxy代理录制实战指南

一、环境准备与配置

1. 安装与启动

# 安装mitmproxy（Python 3.6+）
pip install mitmproxy

# 启动录制代理（端口8080）
mitmdump -w mall_recording.mitm -p 8080

# 查看可用接口
mitmproxy --help

2. 客户端代理配置

二、基础流量录制

1. 简单录制模式

# 录制所有流量到文件
mitmdump -w mall_recording.mitm -p 8080

# 播放录制的流量
mitmdump -n -r mall_recording.mitm -p 8081

2. 过滤指定域名

# 只录制商城相关域名
mitmdump -w mall_recording.mitm -p 8080 --filter "~d mall.com ~d api.mall.com"

三、自动化录制脚本

1. 智能录制脚本

# smart_recorder.py
from mitmproxy import http
import json

class Recorder:
    def __init__(self):
        self.api_count = 0
        self.flows = []

    def request(self, flow: http.HTTPFlow):
        if "api.mall.com" in flow.request.host:
            flow.metadata["start_time"] = time.time()
            self.api_count += 1
            print(f"捕获API请求: {flow.request.method} {flow.request.path}")

    def response(self, flow: http.HTTPFlow):
        if "api.mall.com" in flow.request.host:
            latency = time.time() - flow.metadata["start_time"]
            status = flow.response.status_code
            print(f"API响应: {status} 延迟: {latency:.2f}s")
            self.flows.append({
                "url": flow.request.url,
                "method": flow.request.method,
                "request": flow.request.text,
                "response": flow.response.text,
                "status": status,
                "latency": latency
            })

addons = [Recorder()]

2. 启动带脚本的录制

mitmdump -s smart_recorder.py -w mall_recording.mitm -p 8080

四、流量分析与转换

1. 生成自动化测试用例

# generate_testcases.py
import json
from mitmproxy import io
from mitmproxy.exceptions import FlowReadException

def convert_to_pytest(flow):
    template = f"""
def test_{flow.request.path.replace('/', '_')}():
    url = "{flow.request.url}"
    headers = {dict(flow.request.headers)}
    data = {flow.request.text if flow.request.text else None}
    
    response = requests.{flow.request.method.lower()}(
        url,
        headers=headers,
        json=data if flow.request.method in ["POST", "PUT"] else None
    )
    
    assert response.status_code == {flow.response.status_code}
    assert response.json() == {flow.response.text}
"""
    return template

with open("mall_recording.mitm", "rb") as f:
    flows = []
    reader = io.FlowReader(f)
    try:
        for flow in reader.stream():
            if "api.mall.com" in flow.request.host:
                flows.append(flow)
    except FlowReadException as e:
        print(f"流量文件损坏: {e}")

with open("test_api_auto.py", "w") as f:
    f.write("import requests\n\n")
    for flow in flows[:10]:  # 取前10个生成用例
        f.write(convert_to_pytest(flow))

五、高级录制技巧

1. 接口自动分类

# api_classifier.py
API_CATEGORIES = {
    "user": ["/login", "/register", "/profile"],
    "product": ["/products", "/search", "/detail"],
    "order": ["/cart", "/checkout", "/payment"]
}

def request(self, flow: http.HTTPFlow):
    for category, paths in API_CATEGORIES.items():
        if any(p in flow.request.path for p in paths):
            flow.metadata["category"] = category
            break
    else:
        flow.metadata["category"] = "other"

2. 敏感数据脱敏

# data_scrubber.py
def response(self, flow: http.HTTPFlow):
    if "user" in flow.metadata.get("category", ""):
        import json
        data = json.loads(flow.response.text)
        if "phone" in data:
            data["phone"] = "***" + data["phone"][-4:]
        if "address" in data:
            data["address"] = "<REDACTED>"
        flow.response.text = json.dumps(data)

六、自动化测试集成

1. 结合Pytest框架

# conftest.py
import pytest
from mitmproxy.tools.main import mitmdump
import threading

@pytest.fixture(scope="session")
def proxy():
    def run():
        mitmdump(["-s", "recorder.py", "-w", "test_flows.mitm"])
    
    t = threading.Thread(target=run, daemon=True)
    t.start()
    yield
    # 测试结束后停止代理

@pytest.fixture
def recorded_flows():
    from mitmproxy import io
    flows = []
    with open("test_flows.mitm", "rb") as f:
        reader = io.FlowReader(f)
        try:
            for flow in reader.stream():
                flows.append(flow)
        except:
            pass
    return flows

2. 流量断言测试

# test_replay.py
def test_order_flow(recorded_flows):
    order_flows = [f for f in recorded_flows if "/order/" in f.request.path]
    
    for flow in order_flows:
        response = requests.request(
            flow.request.method,
            flow.request.url,
            headers=dict(flow.request.headers),
            data=flow.request.content
        )
        
        assert response.status_code == flow.response.status_code
        if flow.response.headers.get("Content-Type") == "application/json":
            assert response.json() == flow.response.json()

七、CI/CD集成方案

1. GitHub Actions配置

name: API Test with Mitmproxy

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v2
    - name: Set up Python
      uses: actions/setup-python@v2
    - name: Install dependencies
      run: |
        python -m pip install --upgrade pip
        pip install mitmproxy pytest requests
    - name: Run tests
      run: |
        mitmdump -s recorder.py -w test_flows.mitm &
        pytest tests/ -v
    - name: Upload artifacts
      uses: actions/upload-artifact@v2
      with:
        name: test-flows
        path: test_flows.mitm

八、最佳实践建议

1. 录制策略：按业务场景分开录制（登录流程、下单流程等）每个场景录制3-5次获取典型用例
2. 数据管理：
3. 自动化增强：Response:"""
4. 异常检测：

九、典型问题解决方案

1. HTTPS证书问题

# 启动时跳过证书验证
mitmdump --ssl-insecure -w flows.mitm

# 手动信任证书（手机端）
adb push ~/.mitmproxy/mitmproxy-ca-cert.cer /sdcard/
# 然后在设置中安装证书

2. 录制文件过大

# 按大小自动分割文件
from mitmproxy import ctx

class Rotator:
    def __init__(self):
        self.count = 0
        self.max_size = 100 * 1024 * 1024  # 100MB
    
    def running(self):
        if os.path.getsize(ctx.options.flow_detail) > self.max_size:
            self.count += 1
            new_file = f"recording_{self.count}.mitm"
            os.rename(ctx.options.flow_detail, new_file)

3. 移动端请求捕获不全

# 强制拦截所有流量
def request(flow):
    if flow.request.host not in ["api.mall.com", "static.mall.com"]:
        print(f"拦截非商城流量: {flow.request.host}")
        flow.kill()

通过这套实战方案，您可以快速实现：

1. 商城接口的自动化录制
2. 流量到测试用例的自动转换
3. 异常接口的智能检测
4. 与CI系统的无缝集成
5. 可追溯的接口文档生成

建议从核心交易链路开始实施，逐步扩大覆盖范围，最终建立完整的接口自动化测试体系。