finyx_data_ai/tests/test_parse_sql_result.py

"""
SQL 结果解析接口测试
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch
from app.main import app

client = TestClient(app)


@pytest.fixture
def sample_request_data():
    """示例请求数据"""
    return {
        "file_path": "/tmp/sql_result.xlsx",
        "file_type": "excel",
        "project_id": "project_001"
    }


@pytest.fixture
def mock_parse_result():
    """模拟解析结果"""
    return {
        "tables": [
            {
                "raw_name": "t_user_base_01",
                "display_name": "用户基础信息表",
                "description": "用户基础信息表",
                "fields": [
                    {
                        "raw_name": "user_id",
                        "display_name": "用户ID",
                        "type": "varchar(64)",
                        "comment": "用户的唯一标识符"
                    },
                    {
                        "raw_name": "user_name",
                        "display_name": "用户名",
                        "type": "varchar(50)",
                        "comment": "用户登录名"
                    }
                ],
                "field_count": 2
            }
        ],
        "total_tables": 1,
        "total_fields": 2,
        "parse_time": 0.4,
        "file_info": {
            "file_name": "sql_result.xlsx",
            "file_size": 8192,
            "file_type": "excel"
        }
    }


@pytest.mark.asyncio
async def test_parse_sql_result_success(sample_request_data, mock_parse_result):
    """测试 SQL 结果解析成功"""
    with patch('app.services.parse_sql_result_service.ParseSQLResultService.parse') as mock_parse:
        # 模拟服务返回解析结果
        mock_parse.return_value = mock_parse_result
        
        response = client.post(
            "/api/v1/inventory/parse-sql-result",
            json=sample_request_data
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True
        assert data["code"] == 200
        assert "data" in data
        assert "tables" in data["data"]
        assert len(data["data"]["tables"]) > 0
        assert data["data"]["total_tables"] == 1
        assert data["data"]["total_fields"] == 2
        assert "file_info" in data["data"]


def test_parse_sql_result_request_validation():
    """测试请求验证"""
    # 测试缺少必需字段
    invalid_request = {
        "file_path": "/tmp/sql_result.xlsx"
        # 缺少 project_id
    }
    
    response = client.post(
        "/api/v1/inventory/parse-sql-result",
        json=invalid_request
    )
    
    assert response.status_code == 422  # 验证错误


def test_parse_sql_result_empty_file_path():
    """测试空文件路径"""
    request_data = {
        "file_path": "",
        "project_id": "project_001"
    }
    
    response = client.post(
        "/api/v1/inventory/parse-sql-result",
        json=request_data
    )
    
    assert response.status_code in [422, 400]  # 验证错误


def test_parse_sql_result_with_csv_file():
    """测试 CSV 文件解析"""
    request_data = {
        "file_path": "/tmp/sql_result.csv",
        "file_type": "csv",
        "project_id": "project_001"
    }
    
    mock_result = {
        "tables": [
            {
                "raw_name": "t_order_01",
                "display_name": "订单表",
                "description": "订单表",
                "fields": [
                    {
                        "raw_name": "order_id",
                        "display_name": "订单ID",
                        "type": "varchar(64)",
                        "comment": "订单唯一标识"
                    }
                ],
                "field_count": 1
            }
        ],
        "total_tables": 1,
        "total_fields": 1,
        "parse_time": 0.3,
        "file_info": {
            "file_name": "sql_result.csv",
            "file_size": 4096,
            "file_type": "csv"
        }
    }
    
    with patch('app.services.parse_sql_result_service.ParseSQLResultService.parse') as mock_parse:
        mock_parse.return_value = mock_result
        
        response = client.post(
            "/api/v1/inventory/parse-sql-result",
            json=request_data
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True
        assert data["data"]["total_tables"] == 1


def test_parse_sql_result_auto_detect_file_type():
    """测试自动检测文件类型"""
    request_data = {
        "file_path": "/tmp/sql_result.csv",
        # file_type 不传，应该自动检测
        "project_id": "project_001"
    }
    
    mock_result = {
        "tables": [],
        "total_tables": 0,
        "total_fields": 0,
        "parse_time": 0.2,
        "file_info": {
            "file_name": "sql_result.csv",
            "file_size": 2048,
            "file_type": "csv"
        }
    }
    
    with patch('app.services.parse_sql_result_service.ParseSQLResultService.parse') as mock_parse:
        mock_parse.return_value = mock_result
        
        response = client.post(
            "/api/v1/inventory/parse-sql-result",
            json=request_data
        )
        
        assert response.status_code == 200


def test_parse_sql_result_empty_tables():
    """测试空表列表"""
    request_data = {
        "file_path": "/tmp/empty_result.xlsx",
        "file_type": "excel",
        "project_id": "project_001"
    }
    
    mock_result = {
        "tables": [],
        "total_tables": 0,
        "total_fields": 0,
        "parse_time": 0.1,
        "file_info": {
            "file_name": "empty_result.xlsx",
            "file_size": 1024,
            "file_type": "excel"
        }
    }
    
    with patch('app.services.parse_sql_result_service.ParseSQLResultService.parse') as mock_parse:
        mock_parse.return_value = mock_result
        
        response = client.post(
            "/api/v1/inventory/parse-sql-result",
            json=request_data
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True
        assert data["data"]["total_tables"] == 0


if __name__ == "__main__":
    pytest.main([__file__, "-v"])