LQAgentPlatform/test/construction_review/test_error_codes_pytest.py

"""
施工方案审查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")