在接口自动化测试中，常需按顺序执行多个依赖性HTTP请求（如先登录获取token，再调用业务接口）。为保障流程可控性与错误中断能力，可构建一个支持步骤化、条件终止的请求执行引擎。

核心设计思路：将每个请求步骤抽象为结构化字典，统一调度入口根据请求类型分发至对应方法；执行器以for循环遍历步骤列表，并在任一请求返回非成功状态码时立即退出，避免后续无效调用。

步骤驱动型请求分发

以下方法接收单步配置，识别method字段并路由至内部私有处理函数：

def dispatch_request(self, step_config):
    method = step_config.get("method", "").lower()
    if method == "get":
        return self._execute_get(step_config)
    elif method == "post":
        return self._execute_post(step_config)
    else:
        return {"status": "error", "code": 405, "message": f"Unsupported HTTP method: {method}"}

顺序执行控制器

该方法逐条执行步骤，一旦某步失败（status != "success"），即刻中止流程并返回当前结果：

def run_sequential_steps(self, step_list):
    last_result = None
    for step in step_list:
        last_result = self.dispatch_request(step)
        if last_result.get("status") != "success":
            break
    return last_result

完整实现示例

以下为精简重构后的工具类，移除了冗余注释与调试代码，增强健壮性与可读性：

# -*- coding: utf-8 -*-
import json
import requests
from utils.config_utils import local_config

class SequentialRequestExecutor:
    def __init__(self):
        self.base_url = local_config.HOSTS
        self.session = requests.Session()

    def _execute_get(self, config):
        url = self.base_url + config.get("path", "")
        params = json.loads(config.get("query_params", "{}"))
        headers = json.loads(config.get("headers", "{}")) if config.get("headers") else {}
        
        try:
            resp = self.session.get(url=url, params=params, headers=headers, timeout=10)
            return {
                "status": "success",
                "code": resp.status_code,
                "body": resp.text,
                "headers": dict(resp.headers),
                "elapsed_ms": int(resp.elapsed.total_seconds() * 1000)
            }
        except Exception as e:
            return {"status": "error", "code": 0, "message": str(e)}

    def _execute_post(self, config):
        url = self.base_url + config.get("path", "")
        json_payload = json.loads(config.get("json_body", "{}"))
        headers = json.loads(config.get("headers", "{}")) if config.get("headers") else {}
        
        try:
            resp = self.session.post(
                url=url,
                json=json_payload,
                headers=headers,
                timeout=10
            )
            resp.encoding = resp.apparent_encoding
            return {
                "status": "success",
                "code": resp.status_code,
                "body": resp.text,
                "headers": dict(resp.headers),
                "elapsed_ms": int(resp.elapsed.total_seconds() * 1000)
            }
        except Exception as e:
            return {"status": "error", "code": 0, "message": str(e)}

    def dispatch_request(self, step_config):
        method = step_config.get("method", "").lower()
        if method == "get":
            return self._execute_get(step_config)
        elif method == "post":
            return self._execute_post(step_config)
        else:
            return {"status": "error", "code": 405, "message": f"Method '{method}' not supported"}

    def run_sequential_steps(self, step_list):
        result = None
        for step in step_list:
            result = self.dispatch_request(step)
            if result["status"] != "success":
                break
        return result

# 使用示例
if __name__ == "__main__":
    executor = SequentialRequestExecutor()
    
    steps = [
        {
            "method": "get",
            "path": "/cgi-bin/token",
            "query_params": '{"grant_type":"client_credential","appid":"wx55614004f367f8ca","secret":"65515b46dd758dfdb09420bb7db2c67f"}',
            "headers": '{}'
        },
        {
            "method": "post",
            "path": "/cgi-bin/tags/delete",
            "json_body": '{"tag":{"id":456}}',
            "headers": '{"Content-Type":"application/json"}'
        }
    ]
    
    final_output = executor.run_sequential_steps(steps)
    print(json.dumps(final_output, indent=2, ensure_ascii=False))