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

上次介绍如何《通过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 subprocessimport jsonimport refrom pathlib import Pathfrom typing import Dictclass 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 osimport reimport loggingfrom typing import Dict, Optionalfrom openai import OpenAIlogger = 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 osimport reimport loggingfrom typing import Dict, Optionalfrom openai import OpenAIlogger = 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 osimport sysimport jsonimport loggingfrom pathlib import Pathfrom typing import Dict, Tuple# 导入我们刚才定义的类from generator import ApiTestGeneratorfrom executor import ApiTestExecutorfrom fixer import ApiTestFixerlogging.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 successif __name__ == "__main__":    orchestrator = ApiTestOrchestrator()    success = orchestrator.run()    sys.exit(0 if success else 1)

SKILL.md与转为playwright测试脚本相同