如何在FastAPI单元测试中接收回调并断言returnFlag值

实现思路 #

• 使用FastAPI官方的TestClient调用目标接口
• 启动临时本地HTTP服务器，作为回调地址接收bar发送的POST请求，从中提取returnFlag
• 通过Mock替换subprocess.run的返回值，避免执行真实子进程，保证测试可控性

完整测试代码 #

from fastapi.testclient import TestClient
from unittest.mock import patch
import http.server
import threading
import time
import urllib.parse
from your_main_module import app  # 替换为你的FastAPI应用模块路径

# 全局变量存储回调收到的returnFlag
received_return_flag = None

class CallbackHandler(http.server.BaseHTTPRequestHandler):
    def do_POST(self):
        global received_return_flag
        # 读取请求数据
        content_length = int(self.headers['Content-Length'])
        post_data = self.rfile.read(content_length)
        
        # 解析请求数据（根据bar中实际发送的格式调整，此处以form-data为例）
        data = urllib.parse.parse_qs(post_data.decode('utf-8'))
        received_return_flag = int(data.get('returnFlag')[0])
        
        # 返回成功响应
        self.send_response(200)
        self.end_headers()

def test_xyzservice():
    global received_return_flag
    received_return_flag = None
    
    # 启动临时回调服务器
    server = http.server.HTTPServer(('localhost', 8001), CallbackHandler)
    server_thread = threading.Thread(target=server.serve_forever)
    server_thread.daemon = True
    server_thread.start()
    
    callback_url = "http://localhost:8001/callback"
    
    # Mock子进程调用，指定返回值为1
    with patch('your_main_module.subprocess.run') as mock_run:
        mock_run.return_value.returncode = 1
        
        client = TestClient(app)
        # 发送PUT请求，正确传递请求体和查询参数
        response = client.put(
            f"/xyzservice?callback_url={callback_url}",
            json={"some": "data", "somekey":"somevalue"}
        )
        
        # 断言主接口响应状态
        assert response.status_code == 200
        
        # 等待后台任务执行完成
        time.sleep(0.5)
        
        # 断言回调中的returnFlag符合预期
        assert received_return_flag == 1
        
    # 关闭临时服务器
    server.shutdown()

关键说明 #

• 替换your_main_module为你的FastAPI应用所在的实际模块路径
• 根据bar函数中someDataWithReturnFlag的实际格式，调整CallbackHandler内的数据解析逻辑（如果是JSON格式，改用json.loads解析）
• 用threading启动临时服务器，确保测试过程中能同时处理主接口请求和回调请求
• 通过unittest.mock.patch隔离真实子进程调用，避免测试依赖外部环境

内容的提问来源于stack exchange，提问作者anon