""" 施工方案审查API错误码测试 - pytest版本 使用pytest运行的标准测试套件 """ import pytest import requests import json import uuid import time import os from typing import Dict, Any # pytest fixtures @pytest.fixture(scope="class") def api_config(): """API配置fixture""" return { "base_url": "http://127.0.0.1:8034", "api_prefix": "/sgsc", "valid_user": "user-001", "valid_project_type": "bridge_up_part", "test_callback_url": "http://test.callback.com" } @pytest.fixture def test_file(): """测试文件fixture - 每个测试都创建新的文件对象""" file_path = "data_pipeline/test_rawdata/1f3e1d98-5b4a-4a06-87b3-c7f0413b901a.pdf" class TestFile: def __init__(self): if os.path.exists(file_path): self.file = open(file_path, 'rb') self.file_tuple = (os.path.basename(file_path), self.file, 'application/pdf') self.close_file = True else: self.file = None self.file_tuple = ("test.pdf", b"mock pdf content", "application/pdf") self.close_file = False def get_file(self): """获取文件元组""" if self.close_file and self.file: # 重新打开文件，确保文件未被关闭 self.file.seek(0) return self.file_tuple def cleanup(self): """清理资源""" if self.close_file and self.file: self.file.close() test_file_obj = TestFile() yield test_file_obj test_file_obj.cleanup() class TestFileUploadErrors: """文件上传接口错误码测试""" @pytest.mark.parametrize("test_case,expected_code", [ ("missing_file", "WJSC001"), ("empty_file", "WJSC003"), ("unsupported_format", "WJSC004"), ("invalid_project_type", "WJSC006") ]) def test_file_upload_errors(self, api_config, test_case, expected_code): """测试文件上传各种错误场景""" url = f"{api_config['base_url']}{api_config['api_prefix']}/file_upload" if test_case == "missing_file": # 不上传文件 data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, data=data) elif test_case == "empty_file": # 上传空文件 files = {"file": ("empty.pdf", b"", "application/pdf")} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) elif test_case == "unsupported_format": # 上传不支持的格式 files = {"file": ("test.txt", b"text content", "text/plain")} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) elif test_case == "invalid_project_type": # 无效的工程方案类型 files = {"file": ("test.pdf", b"mock pdf content", "application/pdf")} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": "invalid_type", "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) # 验证错误响应 assert response.status_code in [400, 403, 404] # 允许的业务错误状态码 try: error_data = response.json() assert error_data["code"] == expected_code assert "error_type" in error_data assert "message" in error_data except json.JSONDecodeError: pytest.fail(f"响应不是有效的JSON: {response.text}") def test_file_upload_success(self, api_config, test_file): """测试文件上传成功""" url = f"{api_config['base_url']}{api_config['api_prefix']}/file_upload" files = {"file": test_file.get_file()} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) assert response.status_code == 200 result = response.json() assert "data" in result assert "callback_task_id" in result["data"] assert "id" in result["data"] @pytest.mark.skip(reason="文件大小检查未实现") def test_wjsc005_file_size_exceeded(self, api_config): """测试WJSC005: 文件过大 - 跳过因为未实现""" pass @pytest.mark.skip(reason="认证检查未实现") def test_wjsc007_unauthorized(self, api_config): """测试WJSC007: 认证失败 - 跳过因为未实现""" pass class TestTaskProgressErrors: """进度查询接口错误码测试""" def test_jdlx001_missing_parameters(self, api_config): """测试JDLX001: 请求参数缺失""" url = f"{api_config['base_url']}{api_config['api_prefix']}/task_progress/test-callback-id" response = requests.get(url) # 不提供user参数 assert response.status_code == 400 error_data = response.json() assert error_data["code"] == "JDLX001" @pytest.mark.parametrize("invalid_id", ["short", "123", "invalid-format"]) def test_jdlx002_invalid_param_format(self, api_config, invalid_id): """测试JDLX002: 请求参数格式错误""" url = f"{api_config['base_url']}{api_config['api_prefix']}/task_progress/{invalid_id}" params = {"user": api_config["valid_user"]} response = requests.get(url, params=params) assert response.status_code == 400 error_data = response.json() assert error_data["code"] == "JDLX002" @pytest.mark.parametrize("invalid_user", ["invalid_user", "user-999", ""]) def test_jdlx004_invalid_user(self, api_config, test_file, invalid_user): """测试JDLX004: 用户标识无效""" # 先上传文件获取有效的callback_task_id callback_task_id = self._upload_file_and_get_callback(api_config, test_file) url = f"{api_config['base_url']}{api_config['api_prefix']}/task_progress/{callback_task_id}" params = {"user": invalid_user} response = requests.get(url, params=params) assert response.status_code == 403 error_data = response.json() assert error_data["code"] == "JDLX004" def test_jdlx005_task_not_found(self, api_config): """测试JDLX005: 任务不存在""" fake_callback_id = f"{uuid.uuid4()}-{int(time.time())}" url = f"{api_config['base_url']}{api_config['api_prefix']}/task_progress/{fake_callback_id}" params = {"user": api_config["valid_user"]} response = requests.get(url, params=params) assert response.status_code == 404 error_data = response.json() assert error_data["code"] == "JDLX005" def _upload_file_and_get_callback(self, api_config, test_file): """辅助方法：上传文件并获取callback_task_id""" url = f"{api_config['base_url']}{api_config['api_prefix']}/file_upload" files = {"file": test_file.get_file()} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) assert response.status_code == 200 result = response.json() return result["data"]["callback_task_id"] @pytest.mark.skip(reason="认证检查未实现") def test_jdlx003_unauthorized(self, api_config): """测试JDLX003: 认证失败 - 跳过因为未实现""" pass class TestReviewResultsErrors: """审查结果接口错误码测试""" @pytest.mark.parametrize("invalid_type", ["invalid", "risk", "detail", ""]) def test_scjg001_invalid_type(self, api_config, invalid_type): """测试SCJG001: 结果类型无效""" url = f"{api_config['base_url']}{api_config['api_prefix']}/review_results" payload = { "id": str(uuid.uuid4()), "user": api_config["valid_user"], "type": invalid_type } response = requests.post(url, json=payload) assert response.status_code == 400 error_data = response.json() assert error_data["code"] == "SCJG001" @pytest.mark.parametrize("invalid_id", ["", None]) def test_scjg002_missing_param_id(self, api_config, invalid_id): """测试SCJG002: 缺少文档ID""" url = f"{api_config['base_url']}{api_config['api_prefix']}/review_results" if invalid_id is None: payload = { "user": api_config["valid_user"], "type": "summary" } else: payload = { "id": invalid_id, "user": api_config["valid_user"], "type": "summary" } response = requests.post(url, json=payload) assert response.status_code == 400 error_data = response.json() assert error_data["code"] == "SCJG002" @pytest.mark.parametrize("invalid_format", ["123", "short-id", "invalid-uuid-format"]) def test_scjg003_invalid_id_format(self, api_config, invalid_format): """测试SCJG003: 文档ID格式错误""" url = f"{api_config['base_url']}{api_config['api_prefix']}/review_results" payload = { "id": invalid_format, "user": api_config["valid_user"], "type": "summary" } response = requests.post(url, json=payload) assert response.status_code == 400 error_data = response.json() assert error_data["code"] == "SCJG003" def test_scjg005_invalid_user_review_results(self, api_config, test_file): """测试SCJG005: 用户标识无效（审查结果接口）""" # 先上传文件获取有效的文件ID file_id = self._upload_file_and_get_file_id(api_config, test_file) url = f"{api_config['base_url']}{api_config['api_prefix']}/review_results" payload = { "id": file_id, "user": "invalid_user", "type": "summary" } response = requests.post(url, json=payload) assert response.status_code == 403 error_data = response.json() assert error_data["code"] == "SCJG005" def test_scjg006_task_not_found_review_results(self, api_config): """测试SCJG006: 任务不存在（审查结果接口）""" url = f"{api_config['base_url']}{api_config['api_prefix']}/review_results" payload = { "id": str(uuid.uuid4()), "user": api_config["valid_user"], "type": "summary" } response = requests.post(url, json=payload) assert response.status_code == 404 error_data = response.json() assert error_data["code"] == "SCJG006" def _upload_file_and_get_file_id(self, api_config, test_file): """辅助方法：上传文件并获取文件ID""" url = f"{api_config['base_url']}{api_config['api_prefix']}/file_upload" files = {"file": test_file.get_file()} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) assert response.status_code == 200 result = response.json() return result["data"]["id"] @pytest.mark.skip(reason="认证检查未实现") def test_scjg004_unauthorized(self, api_config): """测试SCJG004: 认证失败 - 跳过因为未实现""" pass class TestIntegration: """集成测试""" def test_complete_workflow_success(self, api_config, test_file): """测试完整工作流程成功场景""" # 1. 文件上传 callback_task_id = self._upload_file_and_get_callback(api_config, test_file) assert callback_task_id is not None # 2. 进度查询 url = f"{api_config['base_url']}{api_config['api_prefix']}/task_progress/{callback_task_id}" params = {"user": api_config["valid_user"]} response = requests.get(url, params=params) assert response.status_code == 200 def _upload_file_and_get_callback(self, api_config, test_file): """辅助方法：上传文件并获取callback_task_id""" url = f"{api_config['base_url']}{api_config['api_prefix']}/file_upload" files = {"file": test_file.get_file()} data = { "callback_url": api_config["test_callback_url"], "project_plan_type": api_config["valid_project_type"], "user": api_config["valid_user"] } response = requests.post(url, files=files, data=data) assert response.status_code == 200 result = response.json() return result["data"]["callback_task_id"] if __name__ == "__main__": # 如果直接运行此文件，给出提示 print("请使用 pytest 运行此测试文件:") print("pytest test/construction_review/test_error_codes_pytest.py -v") print("或者运行所有测试:") print("pytest test/ -v")