首页
学习
活动
专区
圈层
工具
发布
社区首页 >专栏 >通过Python实现需求转为API测试脚本

通过Python实现需求转为API测试脚本

作者头像
顾翔
发布2026-04-28 13:30:46
发布2026-04-28 13:30:46
320
举报

上次介绍如何《通过Python实现需求转为playwright测试脚本》,这次介绍《通过Python实现需求转为API测试脚本》,相对而言转为API测试脚本比转为playwright测试脚本效率要高的很多,只要根据几次生成的结果,调理需求,基本上能生成100%通过的测试脚本,而playwright测试脚本的通过率也可以达到100%,但是概率很小,必须有人工参与调整。

目录

代码语言:javascript
复制

|——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

代码语言:javascript
复制

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

代码语言:javascript
复制

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

代码语言:javascript
复制

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

代码语言:javascript
复制

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测试脚本相同

代码语言:javascript
复制

---
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: 需求文件不存在或格式错误

需求

注册需求

代码语言:javascript
复制

**基本需求**

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表;执行每个测试用例后,请断开数据库连接
- 注册用户成功,请进入数据库中进行检查,注册的数据是否正确存在数据库中

登录需求

代码语言:javascript
复制

**基本需求**
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表信息

找回密码需求

代码语言:javascript
复制

**基本需求**

***输入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

注意

  • 执行main.py的时候将注册需求、登录需求、找回密码需求,定义在名为req.txt文件中
本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2026-04-25,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档