api-testscript-generator

name api-testscript-generator description 接口自动化测试脚本批量生成技能。基于标准化接口定义（api_definitions.json）与可选的全场景测试数据文件，按照团队既定工程规范（Python + Requests + Pytest + Allure2），自动生成分层架构的接口自动化测试脚本。支持数据驱动模式（脚本与数据解耦）和内联数据模式，输出可直接运行的接口自动化脚本工程。 API Test Script Generator - 接口自动化测试脚本批量生成 概述 本技能扮演接口自动化测试架构师角色，核心能力是基于标准化接口定义与可选测试数据，按照团队既定工程规范，批量生成分层架构、数据驱动、可直接运行的接口自动化测试脚本。 两种数据模式： 模式 输入 数据来源 适用场景 数据驱动模式 api_definitions.json + 测试数据目录 外部 YAML/JSON 数据文件 团队规范、数据频繁变更、脚本与数据解耦 内联数据模式 api_definitions.json （无测试数据目录） 脚本内自动生成默认测试数据 快速验证、无测试数据文件、初期搭建 分层架构： api_auto_project/ ├── config/ # 环境配置、全局常量 ├── api/ # 接口请求层（封装所有接口） ├── testcases/ # 测试用例层 ├── data/ # 测试数据（数据驱动模式） ├── utils/ # 工具类 ├── reports/ # 报告输出 ├── conftest.py # Pytest 全局钩子 └── pytest.ini # Pytest 配置 触发条件 以下场景自动触发本技能： 用户提供 api_definitions.json 或 api_definitions.yaml 文件，要求生成接口自动化脚本 用户同时提供 api_definitions.json + 测试数据目录，要求按数据驱动方式生成脚本 用户要求"生成接口测试脚本""生成自动化脚本""接口自动化代码生成" 用户提及"api-testscript-generator""/api_testscript_generator" 用户需要将接口定义转化为可执行的 Pytest 自动化测试工程 用户需要基于 api-testdata-generator 的输出结果生成对应的测试脚本 输入 必需输入 标准化接口定义文件 ：由 api-schema-parser 输出的 api_definitions.json/yaml 包含 meta 、 apis 、 global_rules 顶层结构 每个接口含 api_id 、 path 、 method 、 parameters 、 responses 、 business_rules 可选输入 全场景测试数据目录 ：由 api-testdata-generator 输出的测试数据文件目录 每个接口对应一个 YAML/JSON 数据文件 数据文件包含 test_cases 列表，每条含 case_id 、 name 、 category 、 parameters 、 expected 目录按模块分组织（如 auth/ 、 order/ ） 目标接口/模块筛选 （选填）：接口 api_id 或模块名称，仅生成指定范围的脚本 自定义输出路径 （选填）：指定生成脚本的根目录，默认为当前目录下的 api_auto_project/ 执行流程 输入（api_definitions.json + 可选测试数据目录） ↓ Step 1: 读取接口结构与参数约束 ↓ Step 2: 识别数据模式（数据驱动 / 内联数据） ↓ Step 3: 生成项目基础设施（config/、utils/、conftest.py、pytest.ini） ↓ Step 4: 生成接口请求层（api/ 层封装） ↓ Step 5: 生成测试数据层（data/ 层，数据驱动模式） ↓ Step 6: 生成测试用例层（testcases/ 层） ↓ Step 7: 注入健壮逻辑（超时、重试、异常捕获、日志） ↓ Step 8: 输出校验与汇总 Step 1: 读取接口结构与参数约束 从 api_definitions.json 中逐个读取接口定义，提取以下关键信息： 信息类型 来源字段 用途 接口路径 path 拼接请求 URL 请求方法 method 确定请求类型（GET/POST/PUT/DELETE） 路径参数 parameters.path_params URL 路径替换 查询参数 parameters.query_params 拼接 Query String 请求头 parameters.header_params 构建 Headers 请求体 parameters.body_params 构建 Request Body 响应结构 responses.success / responses.errors 生成断言 业务规则 business_rules 鉴权、幂等等特殊处理 全局规则 global_rules 全局鉴权、限流等 模块归属 module 文件与目录组织 Step 2: 识别数据模式 判定条件 数据模式 处理方式 用户提供了测试数据目录，且目录下存在对应的 YAML/JSON 文件 数据驱动模式 将数据文件映射到 data/ 层，用例层通过 @pytest.mark.parametrize 或 yaml.safe_load() 读取 用户未提供测试数据目录，或数据目录为空 内联数据模式 在 testcases/ 层用例中直接构建测试数据字典，或生成简单的 data/ 层默认数据 数据文件匹配规则（数据驱动模式）： 测试数据目录结构 对应接口 匹配方式 auth/user_login.yaml POST_/api/auth/login 按模块/文件名映射 order/create_order.yaml POST_/api/order/create 按模块/文件名映射 匹配策略：优先按 api_id 精确匹配，其次按接口名称模糊匹配，最后按模块+路径推断。 Step 3: 生成项目基础设施 3.1 config/ 环境配置 config/config.py ： """ 全局配置模块 - 环境切换：dev/test/pre/prod - 读取对应环境配置文件 """ import os import yaml BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) ENV = os.getenv( "API_TEST_ENV" , "test" ) def load_config (): """加载环境配置""" config_path = os.path.join(BASE_DIR, "config" , f" {ENV} .yaml" ) with open (config_path, "r" , encoding= "utf-8" ) as f: return yaml.safe_load(f) CONFIG = load_config() # 全局常量 BASE_URL = CONFIG.get( "base_url" , "" ) TIMEOUT = CONFIG.get( "timeout" , 30 ) MAX_RETRY = CONFIG.get( "max_retry" , 2 ) CONTENT_TYPE = "application/json" config/dev.yaml / config/test.yaml ： # test.yaml 示例 base_url: "https://test-api.example.com" timeout: 30 max_retry: 2 auth: login_url: "/api/auth/login" username: "testuser" password: "Test@1234" database: host: "test-db.example.com" port: 3306 name: "test_db" 3.2 utils/ 工具类 utils/logger.py ： """ 统一日志模块 - 格式：时间 - 级别 - 模块 - 信息 - 请求/响应自动打印 """ import logging import os LOG_DIR = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "reports" , "logs" ) os.makedirs(LOG_DIR, exist_ok= True ) def get_logger ( name ): """获取 Logger 实例""" logger = logging.getLogger(name) if not logger.handlers: logger.setLevel(logging.DEBUG) formatter = logging.Formatter( "%(asctime)s - %(levelname)s - %(name)s - %(message)s" ) # 控制台输出 console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_handler.setFormatter(formatter) logger.addHandler(console_handler) # 文件输出 file_handler = logging.FileHandler( os.path.join(LOG_DIR, f" {name} .log" ), encoding= "utf-8" ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger utils/request_util.py ： """ 统一请求工具类 - 超时统一 30s - 失败自动重试 2 次 - 异常捕获（连接超时/读取超时/连接错误/代理异常/数据解析异常） - 请求/响应自动日志 - Allure 步骤记录 """ import allure import requests from requests.exceptions import ( ConnectionError, ProxyError, ReadTimeout, ConnectTimeout, JSONDecodeError, ) from utils.logger import get_logger from config.config import TIMEOUT, MAX_RETRY logger = get_logger( "request_util" ) class RequestUtil : """统一请求封装""" def __init__ ( self, session= None ): self .session = session or requests.Session() def _retry_request ( self, method, url, **kwargs ): """带重试的请求方法""" kwargs.setdefault( "timeout" , TIMEOUT) last_exception = None for attempt in range ( 1 , MAX_RETRY + 2 ): # 首次 + MAX_RETRY 次重试 try : logger.info( f"[Request] {method} {url} | attempt= {attempt} " ) response = self .session.request(method, url, **kwargs) logger.info( f"[Response] status= {response.status_code} | " f"time= {response.elapsed.total_seconds(): .3 f} s" ) return response except (ConnectTimeout, ReadTimeout) as e: last_exception = e logger.warning( f"[Retry] Timeout on attempt {attempt} : {e} " ) except ConnectionError as e: last_exception = e logger.warning( f"[Retry] ConnectionError on attempt {attempt} : {e} " ) except ProxyError as e: last_exception = e logger.warning( f"[Retry] ProxyError on attempt {attempt} : {e} " ) except JSONDecodeError as e: last_exception = e logger.error( f"[Error] JSONDecodeError: {e} " ) raise except Exception as e: last_exception = e logger.error( f"[Error] Unexpected: {e} " ) raise raise last_exception @allure.step( "GET {url}" ) def get ( self, url, **kwargs ): return self ._retry_request( "GET" , url, **kwargs) @allure.step( "POST {url}" ) def post ( self, url, **kwargs ): return self ._retry_request( "POST" , url, **kwargs) @allure.step( "PUT {url}" ) def put ( self, url, **kwargs ): return self ._retry_request( "PUT" , url, **kwargs) @allure.step( "DELETE {url}" ) def delete ( self, url, **kwargs ): return self ._retry_request( "DELETE" , url, **kwargs) @allure.step( "PATCH {url}" ) def patch ( self, url, **kwargs ): return self ._retry_request( "PATCH" , url, **kwargs) utils/assert_util.py ： """ 统一断言工具类 - 三层断言：状态码 + 业务码 + 业务数据 - Allure 步骤记录 - 断言失败自动附加响应信息 """ import allure from utils.logger import get_logger logger = get_logger( "assert_util" ) class AssertUtil : """统一断言封装""" @staticmethod @allure.step( "断言状态码" ) def assert_status_code ( response, expected_code ): """断言 HTTP 状态码""" actual_code = response.status_code assert actual_code == expected_code, ( f"状态码断言失败: 期望= {expected_code} , 实际= {actual_code} | " f"响应= {response.text[: 500 ]} " ) @staticmethod @allure.step( "断言业务码" ) def assert_business_code ( response, expected_code ): """断言业务状态码""" try : json_data = response.json() except Exception: assert False , f"响应非JSON格式: {response.text[: 500 ]} " actual_code = json_data.get( "code" ) assert actual_code == expected_code, ( f"业务码断言失败: 期望= {expected_code} , 实际= {actual_code} | " f"message= {json_data.get( 'message' , '' )} | " f"响应= {response.text[: 500 ]} " ) @staticmethod @allure.step( "断言业务数据" ) def assert_business_data ( response, field_checks ): """ 断言业务数据 field_checks: list of dict [{"field": "data.id", "check": "not_empty"}, {"field": "data.username", "check": "equals", "expect": "zhangsan"}] """ try : json_data = response.json() except Exception: assert False , f"响应非JSON格式: {response.text[: 500 ]} " for check_item in field_checks: field_path = check_item[ "field" ] check_type = check_item[ "check" ] # 按路径取值 value = json_data for key in field_path.split( "." ): if isinstance (value, dict ): value = value.get(key) elif isinstance (value, list ) and key.isdigit(): value = value[ int (key)] else : value = None break if check_type == "not_empty" : assert value is not None and value != "" , ( f"字段非空断言失败: {field_path} 值为空 | 响应= {response.text[: 500 ]} " ) elif check_type == "equals" : expect = check_item[ "expect" ] assert value == expect, ( f"字段匹配断言失败: {field_path} 期望= {expect} , 实际= {value} | " f"响应= {response.text[: 500 ]} " ) elif check_type == "type" :