finyx_data_ai/tests/test_parse_document.py

"""
文档解析接口测试
"""
import pytest
from fastapi.testclient import TestClient
from unittest.mock import patch, AsyncMock
from app.main import app

client = TestClient(app)


@pytest.fixture
def sample_request_data():
    """示例请求数据"""
    return {
        "file_path": "/tmp/test_document.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": "从 Excel Sheet 't_user_base_01' 解析",
                "fields": [
                    {
                        "raw_name": "user_id",
                        "display_name": "user_id",
                        "type": "varchar(255)",
                        "comment": None,
                        "is_primary_key": False,
                        "is_nullable": True,
                        "default_value": None
                    },
                    {
                        "raw_name": "user_name",
                        "display_name": "user_name",
                        "type": "varchar(255)",
                        "comment": None,
                        "is_primary_key": False,
                        "is_nullable": True,
                        "default_value": None
                    }
                ],
                "field_count": 2
            }
        ],
        "total_tables": 1,
        "total_fields": 2,
        "parse_time": 0.5,
        "file_info": {
            "file_name": "test_document.xlsx",
            "file_size": 10240,
            "file_type": "excel"
        }
    }


@pytest.mark.asyncio
async def test_parse_document_success(sample_request_data, mock_parse_result):
    """测试文档解析成功"""
    with patch('app.services.parse_document_service.ParseDocumentService.parse') as mock_parse:
        # 模拟服务返回解析结果
        mock_parse.return_value = mock_parse_result
        
        response = client.post(
            "/api/v1/inventory/parse-document",
            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_document_request_validation():
    """测试请求验证"""
    # 测试缺少必需字段
    invalid_request = {
        "file_path": "/tmp/test.xlsx"
        # 缺少 project_id
    }
    
    response = client.post(
        "/api/v1/inventory/parse-document",
        json=invalid_request
    )
    
    assert response.status_code == 422  # 验证错误


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


def test_parse_document_with_word_file():
    """测试 Word 文件解析"""
    request_data = {
        "file_path": "/tmp/test_document.docx",
        "file_type": "word",
        "project_id": "project_001"
    }
    
    mock_result = {
        "tables": [],
        "total_tables": 0,
        "total_fields": 0,
        "parse_time": 0.3,
        "file_info": {
            "file_name": "test_document.docx",
            "file_size": 5120,
            "file_type": "word"
        }
    }
    
    with patch('app.services.parse_document_service.ParseDocumentService.parse') as mock_parse:
        mock_parse.return_value = mock_result
        
        response = client.post(
            "/api/v1/inventory/parse-document",
            json=request_data
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True


def test_parse_document_with_pdf_file():
    """测试 PDF 文件解析"""
    request_data = {
        "file_path": "/tmp/test_document.pdf",
        "file_type": "pdf",
        "project_id": "project_001"
    }
    
    mock_result = {
        "tables": [],
        "total_tables": 0,
        "total_fields": 0,
        "parse_time": 1.0,
        "file_info": {
            "file_name": "test_document.pdf",
            "file_size": 20480,
            "file_type": "pdf"
        }
    }
    
    with patch('app.services.parse_document_service.ParseDocumentService.parse') as mock_parse:
        mock_parse.return_value = mock_result
        
        response = client.post(
            "/api/v1/inventory/parse-document",
            json=request_data
        )
        
        assert response.status_code == 200
        data = response.json()
        assert data["success"] is True


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


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