上次介绍如何《通过Python实现需求转为playwright测试脚本》,这次介绍《通过Python实现需求转为API测试脚本》,相对而言转为API测试脚本比转为playwright测试脚本效率要高的很多,只要根据几次生成的结果,调理需求,基本上能生成100%通过的测试脚本,而playwright测试脚本的通过率也可以达到100%,但是概率很小,必须有人工参与调整。
目录
|——logs
| |
| ———test_run.log(运行后自动生成)
|——outputs
| |
| ——— /test_history (目录、运行后自动生成)
| test_api_current.py (运行后生成的测试程序)
| fix_report.json (运行后自动生成)
| report.json (运行后自动生成)
|
|——skills
| |
|———/test-orchestrator
| |
|——————skill.md: 能力文件
| /scripts
| |
|—————————executor.py :运行测试程序
| fixer.py:修复测试程序
| generator.py:生成测试程序
| orchestrator.py:调度者
|——main.py 主程序
|——req.txt 需求文档
实现代码
executor.py
import subprocess
import json
import re
from pathlib import Path
from typing import Dict
class ApiTestExecutor:
def __init__(self):
# 接口测试不需要 headless 参数
pass
def run(self, test_file: str) -> Dict:
"""执行测试文件"""
test_path = Path(test_file)
if not test_path.exists():
return {"success": False, "error": "文件不存在", "exit_code": -1}
# 构建 pytest 命令
cmd = [
"pytest", str(test_path),
"-v", "--tb=short", "--color=no",
"--json-report", f"--json-report-file={test_path.parent}/report.json"
]
try:
# 修复编码问题:设置 encoding='utf-8' 和 errors='replace'
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=60,
encoding='utf-8', # 强制使用 UTF-8 编码
errors='replace', # 替换无法解码的字符
env={**subprocess.os.environ, "PYTHONUNBUFFERED": "1", "PYTHONIOENCODING": "utf-8"}
)
# 修复 None 值问题:如果 stdout 或 stderr 为 None,替换为空字符串
stdout = result.stdout if result.stdout is not None else ""
stderr = result.stderr if result.stderr is not None else ""
# 解析结果
passed = []
failed = []
for line in stdout.split("\n"):
if "PASSED" in line:
passed.append(line)
elif "FAILED" in line:
failed.append(line)
error_msg = stdout + "\n" + stderr
# 尝试读取 json 报告获取更详细的错误
report_file = test_path.parent / "report.json"
if report_file.exists():
try:
with open(report_file, 'r', encoding='utf-8') as f:
report = json.load(f)
# 提取具体的失败信息
for test in report.get("tests", []):
if test.get("outcome") == "failed":
error_msg = test.get("call", {}).get("crash", {}).get("message", error_msg)
except Exception as e:
error_msg += f"\n读取报告失败: {str(e)}"
return {
"success": result.returncode == 0,
"passed": passed,
"failed": failed,
"error": error_msg[:2000],
"exit_code": result.returncode
}
except subprocess.TimeoutExpired:
return {"success": False, "error": "执行超时", "exit_code": -2}
except UnicodeDecodeError as e:
return {"success": False, "error": f"编码错误: {str(e)}", "exit_code": -3}
except Exception as e:
return {"success": False, "error": f"执行异常: {str(e)}", "exit_code": -4}
fixer.py
import os
import re
import logging
from typing import Dict, Optional
from openai import OpenAI
logger = logging.getLogger(__name__)
class ApiTestFixer:
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
self.model = os.getenv("QWEN_MODEL", "qwen-plus")
self.max_retries = 3
def _classify_error(self, error_msg: str) -> Dict:
"""针对接口测试的错误分类"""
error_lower = error_msg.lower()
if "404" in error_msg:
return {"type": "Http404", "severity": "high"}
elif "500" in error_msg:
return {"type": "Http500", "severity": "critical"}
elif "401" in error_msg or "403" in error_msg:
return {"type": "AuthError", "severity": "high"}
elif "json" in error_lower and "decode" in error_lower:
return {"type": "JsonDecodeError", "severity": "medium"}
elif "assert" in error_lower:
return {"type": "BusinessLogicAssertion", "severity": "high"}
else:
return {"type": "Unknown", "severity": "low"}
def _extract_code(self, content: str) -> Optional[str]:
# 同之前的逻辑
if "```python" in content:
start = content.find("```python") + 9
end = content.find("```", start)
return content[start:end].strip()
return content.strip()
def fix(self, code: str, error_info: Dict, requirements: str) -> str:
error_detail = error_info.get("error", "")
error_type = self._classify_error(error_detail)
prompt = f"""
你是一个接口自动化测试专家。请修复以下基于 `requests` 的测试代码。
## 错误分析
- 类型: {error_type['type']}
- 详情: {error_detail}
## 需求
{requirements}
## 失败代码
```python
{code}
```
## 修复指南
1. **URL 检查**:如果是 404,检查 BASE_URL 和路由拼接是否正确。
2. **参数格式**:检查 `json=payload` 还是 `data=payload`,确保 Content-Type 匹配。
3. **依赖关系**:如果是 401/403,检查是否缺少 Token 或 Cookie,是否需要先调用登录接口。
4. **断言修复**:根据错误信息调整断言逻辑,确保标点符号与需求一致。
5. **JSON 解析**:如果报错 JSON decode,先打印 response.text 再尝试解析。
请输出修复后的完整代码。
"""
for _ in range(self.max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1
)
content = response.choices[0].message.content
fixed_code = self._extract_code(content)
if fixed_code and "import requests" in fixed_code:
return fixed_code
except Exception as e:
logger.error(f"修复失败: {e}")
return code # 失败返回原代码```
generator.py
import os
import re
import logging
from typing import Dict, Optional
from openai import OpenAI
logger = logging.getLogger(__name__)
class ApiTestGenerator:
def __init__(self):
self.client = OpenAI(
api_key=os.getenv("DASHSCOPE_API_KEY"),
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1"
)
self.model = os.getenv("QWEN_MODEL", "qwen-plus")
def _extract_code(self, content: str) -> str:
"""提取代码块"""
if "```python" in content:
start = content.find("```python") + 9
end = content.find("```", start)
return content[start:end].strip()
return content.strip()
def generate(self, requirements: str) -> str:
"""生成基于 Requests 的接口测试代码"""
prompt = f"""
你是一个 Python 接口自动化测试专家。请根据以下产品需求,编写基于 `requests` 和 `pytest` 的测试代码。
## 产品需求
{requirements}
## 核心规范
1. **库的使用**:
- 必须使用 `import requests`。
- 使用 `pytest` 进行断言和组织。
- 使用 `parameterized` 进行数据驱动测试。
2. **会话管理**:
- 使用 `requests.Session()` 来保持 Cookie 或 Header。
- 在 `@pytest.fixture` 中初始化 Session。
3. **URL 管理**:
- 定义 `BASE_URL = "http://localhost:8080/api"` (根据需求推断)。
4. **断言策略**:
- 优先断言 HTTP 状态码:`assert response.status_code == 200`。
- 断言业务逻辑:`assert response.json().get("code") == 0` 或 `assert "success" in response.text`。
- **标点符号保护**:如果需求中规定了错误消息(如“用户名不存在!”),断言时必须包含标点符号。
5. **数据库清理**:
- 如果需要数据库验证,保留 pymysql 连接代码,但在测试结束后清理数据。
## 代码模板参考
```python
import pytest
import requests
from parameterized import parameterized
BASE_URL = "http://localhost:8080/api"
class TestUserAPI:
@pytest.fixture(autouse=True)
def setup(self):
self.session = requests.Session()
self.session.headers.update({{"Content-Type": "application/json"}})
yield
self.session.close()
@parameterized.expand([
("正常登录", "admin", "123456", 200, "success"),
("密码错误", "admin", "wrong", 401, "密码错误"),
])
def test_login(self, name, user, pwd, exp_status, exp_msg):
url = f"{{BASE_URL}}/login"
payload = {{"username": user, "password": pwd}}
response = self.session.post(url, json=payload)
assert response.status_code == exp_status
json_data = response.json()
assert exp_msg in json_data.get("message", "")
```
请输出完整的 Python 代码,不要解释。
"""
response = self.client.chat.completions.create(
model=self.model,
messages=[{"role": "user", "content": prompt}],
temperature=0.1,
max_tokens=4000
)
code = self._extract_code(response.choices[0].message.content)
# 确保导入了必要的库
if "import requests" not in code:
code = "import requests\n" + code
if "from parameterized import parameterized" not in code:
code = "from parameterized import parameterized\n" + code
return code
orchestrator.py
import os
import sys
import json
import logging
from pathlib import Path
from typing import Dict, Tuple
# 导入我们刚才定义的类
from generator import ApiTestGenerator
from executor import ApiTestExecutor
from fixer import ApiTestFixer
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ApiTestOrchestrator:
def __init__(self, req_file="req.txt", max_retry=5, output_dir="outputs"):
self.req_file = Path(req_file)
self.max_retry = max_retry
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
self.current_test_file = self.output_dir / "test_api_current.py"
# 初始化组件
self.generator = ApiTestGenerator()
self.executor = ApiTestExecutor()
self.fixer = ApiTestFixer()
def run(self) -> bool:
if not self.req_file.exists():
logger.error(f"需求文件不存在: {self.req_file}")
return False
# 读取需求时指定编码
requirements = self.req_file.read_text(encoding="utf-8")
logger.info(f"读取需求: {requirements[:50]}...")
code = None
success = False
last_result = None # 初始化 last_result
for i in range(self.max_retry):
logger.info(f"\n--- 第 {i+1} 轮迭代 ---")
# 1. 生成或修复
if i == 0:
logger.info("正在生成接口测试代码...")
code = self.generator.generate(requirements)
else:
logger.info("正在修复代码...")
code = self.fixer.fix(code, last_result, requirements)
# 保存代码时指定 UTF-8 编码
self.current_test_file.write_text(code, encoding="utf-8")
logger.info(f"代码已保存至: {self.current_test_file}")
# 2. 执行
logger.info("正在执行测试...")
result = self.executor.run(str(self.current_test_file))
last_result = result
# 3. 检查结果
if result["success"]:
logger.info("测试全部通过!")
success = True
break
else:
logger.warning(f"测试失败: {result['error'][:200]}")
return success
if __name__ == "__main__":
orchestrator = ApiTestOrchestrator()
success = orchestrator.run()
sys.exit(0 if success else 1)
SKILL.md 与转为playwright测试脚本相同
---
name: api-test-orchestrator
description: |
基于 Requests 的接口自动化测试编排器。
读取需求 → 生成 API 测试 → 执行 → 失败自动修复 → 重试,直至全部通过。
version: 1.0.0
author: AI Agent
---
# 接口测试编排技能
## 能力概述
本技能实现了一个轻量级、高效率的接口测试自动化闭环流程:
1. **需求解析**:从 `req.txt` 读取产品需求文档
2. **代码生成**:调用 LLM 基于 `requests` + `pytest` 生成接口测试代码(含 Session 管理、参数化)
3. **自动执行**:在本地环境运行 Pytest
4. **智能修复**:如果测试失败,分析 HTTP 状态码、JSON 响应及断言错误,调用 LLM 修复代码
5. **闭环重试**:重复执行与修复步骤,直到所有用例通过或达到最大重试次数(默认 5 次)
## 输入参数
- `--req_file`: 需求文件路径,默认 `req.txt`
- `--max_retry`: 最大修复重试次数,默认 `5`
- `--verbose`: 详细日志输出,默认 `false`
## 输出产物
- `outputs/test_api_current.py`: 最终通过的接口测试代码
- `outputs/fix_report.json`: 修复过程记录(包含错误类型、修复次数)
- `logs/test_run.log`: 完整运行日志
## 编排流程
```mermaid
graph TD
A[读取 req.txt] --> B[LLM 生成 requests 代码]
B --> C[执行 Pytest]
C --> D{测试通过?}
D -- 是 --> E[完成!输出最终代码]
D -- 否 --> F[分析错误日志]
F --> G[LLM 修复代码]
G --> C
style E fill:#9f9,stroke:#333,stroke-width:2px
style C fill:#ff9,stroke:#333,stroke-width:2px
##错误处理策略
- HTTP 404/500:自动检查 URL 拼接、Header 设置及服务端异常处理
- JSON 解析错误:自动添加 response.text 打印调试,并优化解析逻辑
- 业务断言失败:根据需求文档修正断言字段及标点符号匹配
##退出码
- 0: 所有测试通过
- 1: 达到最大重试次数仍有失败
- 2: 需求文件不存在或格式错误
需求
注册需求
**基本需求**
request.post(url,data,cookies)
url = http://127.0.0.1:8080/ChatGPTEbusiness/jsp/RegisterPage.jsp
data = { "csrftoken"=csrftoken,
"username"=username,
"password"=password,
"phone"=phone,
"email"=email
}
其中password经过SHA256散列
cookies = {"csrftoken"=csrftoken}
csrftoken来自id="csrftoken" input的value值,即<input type="hidden" id="csrftoken" name="csrftoken" value="DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE">
中的"DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE
**注意**:每一次请求都要获取一次csrftoken
**信息**
- 注册成功:登录页面
- 账号(必填):文本框,长度为5-20位,可以包含大小写英文字符(必填)或数字(选填)
正则表达式 "^[a-zA-Z0-9]{5,20}$"。
错误信息:"账号必须是5-20位字母或数字"
- 手机号(必填):手机框,需符合中国手机号码格式。
正则表达式 "^1[3-9]\\d{9}$"。
错误信息:"手机号必须符合中国手机号码格式"
- Email(必填):需符合国际标准Email格式。
正则表达式 "^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\\.[a-zA-Z0-9-.]+$"。
错误信息:"Email格式不正确"
- 密码没有进行SHA256散列
错误信息:"密码应该哈希进行存储"
- cookies中的csrftoken与data中的csrftoken不一致
错误信息:"可能存在CSRF注入风险"
**数据库信息**
## 测试环境配置
```
DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': '123456',
'database': 'chatgptebusiness'
}
```
## user表格式
```
CREATE TABLE IF NOT EXISTS user(
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL,
password VARCHAR(100) NOT NULL,
phone VARCHAR(50) NOT NULL,
email VARCHAR(50) NOT NULL
);
```
- 请执行每个测试用例前,建立数据库连接,清除user表;执行每个测试用例后,请断开数据库连接
- 注册用户成功,请进入数据库中进行检查,注册的数据是否正确存在数据库中
登录需求
**基本需求**
request.post(url,data,cookies)
url = http://127.0.0.1:8080/ChatGPTEbusiness/jsp/LoginPage.jsp
data = { "csrftoken"=csrftoken,
"username"=username,
"password"=password
}
其中password经过SHA256散列
cookies = {"csrftoken"=csrftoken}
csrftoken来自id="csrftoken" input的value值,即<input type="hidden" id="csrftoken" name="csrftoken" value="DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE">
中的"DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE
**注意**:每一次请求都要获取一次csrftoken
**信息**
- 登录成功:系统欢迎您
- 账号(必填):文本框,长度为5-20位,可以包含大小写英文字符(必填)或数字(选填)
正则表达式 "^[a-zA-Z0-9]{5,20}$"。
错误信息:"账号必须是5-20位字母或数字"
- 密码没有进行SHA256散列
错误信息:"密码应该哈希进行存储"
- cookies中的csrftoken与data中的csrftoken不一致
错误信息:"可能存在CSRF注入风险"
**数据库信息**
## 测试环境配置
```
DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': '123456',
'database': 'chatgptebusiness'
}
```
## user表格式
```
CREATE TABLE IF NOT EXISTS user(
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL,
password VARCHAR(100) NOT NULL,
phone VARCHAR(50) NOT NULL,
email VARCHAR(50) NOT NULL
);
```
- 请执行每个测试用例前,建立数据库连接,建立准备登录的user表信息;执行每个测试用例后,请断开数据库连接,删除建立的user表信息
找回密码需求
**基本需求**
***输入phone或Email***
- request.post(url,data,cookies)
- url = http://127.0.0.1:8080/ChatGPTEbusiness/jsp/VeriCodePage.jsp
- data = {"csrftoken"=csrftoken,
"contact"=contact}
- cookies = {"csrftoken"=csrftoken}
- csrftoken来自VeriCodePage.jsp的id="csrftoken" input的value值,即<input type="hidden" id="csrftoken" name="csrftoken" value="DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE">
中的"DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE
**注意**:
- 每一次请求都要获取一次csrftoken
- contact均使用一个有效的Email地址: xianggu625@126.com
***找回密码***
- request.post(url,data,cookies)
- url = http://127.0.0.1:8080/ChatGPTEbusiness/jsp/RecoverPage.jsp
- data = {"csrftoken"=csrftoken,
"identifyingCode"=identifyingCode,
"newPassword"=newPassword
}
newPassword需要SHA256散列
- cookies = {"csrftoken"=csrftoken}
- csrftoken来自RecoverPage.jsp的id="csrftoken" input的value值,即<input type="hidden" id="csrftoken" name="csrftoken" value="DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE">
中的"DlFaArVGOOMBgmPFENFH8s7mD7v2c8rtNfiOBosjUC5QCqkPcVBSlgeyhGFqSFEDUlaGHmx5FiL8uA4hnMNX0NvgZDpuSocr2wqE
**注意**:
- 每一次请求都要获取一次csrftoken
- 操作RecoverPage.jsp前必须先操作VeriCodePage.jsp,不能一上来就操作RecoverPage.jsp
- identifyingCode必须从数据库表code获取,然后在data中进行传输,不得在测试程序中向数据库中插入数据。
**信息**
- 输入phone或Email成功:"找回密码"。
- 重置密码:"登录页面"。
- 手机号或Email在user表中查不到:"您输入的手机号或Email不存在,请重新输入!"(这个测试仅在VeriCodePage.jsp页面测试即可,不用在RecoverPage.jsp)
- 输入的验证码与code表中的验证码不一致:"验证码错误,请重新输入!"。
- 新密码以前使用过:"这个密码以前设置过,请用一个新密码!"。
- 新密码没有SHA56散列:"密码需要HASH散列"
**数据库信息**
```
DB_CONFIG = {
'host': 'localhost',
'user': 'root',
'password': '123456',
'database': 'chatgptebusiness'
}
```
***user表格式***
```
CREATE TABLE IF NOT EXISTS user(
id INT AUTO_INCREMENT PRIMARY KEY,
username VARCHAR(50) NOT NULL,
password VARCHAR(100) NOT NULL,
phone VARCHAR(50) NOT NULL,
email VARCHAR(50) NOT NULL
);
```
***code表格式***
```
CREATE TABLE code(
id INT AUTO_INCREMENT PRIMARY KEY,
uid INT NOT NULL,
code CHAR(6) NOT NULL,
FOREIGN KEY(uid) REFERENCES user(id)
);
```
***password表格式***
```
CREATE TABLE password(
id INT AUTO_INCREMENT PRIMARY KEY,
uid INT NOT NULL,
password VARCHAR(100) NOT NULL,
FOREIGN KEY(uid) REFERENCES user(id)
);
```
- 完成每个测试用例前请删除user表、code表和password表.
- 执行每一个用例前建立一个user信息{"jerrygu","Zxcv@123","13681732596","xianggu625@126.com"}
**URL**
- 输入phone或Email:http://127.0.0.1:8080/ChatGPTEbusiness/jsp/VeriCodePage.jsp
- 重置密码:http://127.0.0.1:8080/ChatGPTEbusiness/jsp/RecoverPage.jsp
注意