cma-management/backend/tests/test_data.py

"""数据管理模块测试 — Excel导入/导出/数据源CRUD"""
import io
import pandas as pd
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.orm import Session
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi


class TestDataSources:
    """数据源CRUD测试"""

    DATA_ENDPOINT = "/api/cma/data"

    def test_list_sources_empty(self, client: TestClient, db: Session):
        """空列表"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get(f"{self.DATA_ENDPOINT}/sources", headers=auth_header(token))
        assert resp.status_code == 200
        assert resp.json()["data"] == []

    def test_create_source(self, client: TestClient, db: Session):
        """创建数据源"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.post(
            f"{self.DATA_ENDPOINT}/sources",
            headers=auth_header(token),
            json={"name": "ERP数据源", "source_type": "erp", "api_endpoint": "http://erp.example.com"},
        )
        assert resp.status_code == 200
        data = resp.json()["data"]
        assert data["name"] == "ERP数据源"
        assert data["source_type"] == "erp"
        assert data["status"] == "active"

    def test_update_source(self, client: TestClient, db: Session):
        """更新数据源"""
        create_test_user(db)
        token = get_token_for_user(client)

        # 先创建
        create_resp = client.post(
            f"{self.DATA_ENDPOINT}/sources",
            headers=auth_header(token),
            json={"name": "旧名称", "source_type": "manual"},
        )
        source_id = create_resp.json()["data"]["id"]

        # 更新
        resp = client.put(
            f"{self.DATA_ENDPOINT}/sources/{source_id}",
            headers=auth_header(token),
            json={"name": "新名称"},
        )
        assert resp.status_code == 200
        assert resp.json()["data"]["name"] == "新名称"

    def test_delete_source(self, client: TestClient, db: Session):
        """删除数据源"""
        create_test_user(db)
        token = get_token_for_user(client)

        create_resp = client.post(
            f"{self.DATA_ENDPOINT}/sources",
            headers=auth_header(token),
            json={"name": "待删除", "source_type": "manual"},
        )
        source_id = create_resp.json()["data"]["id"]

        resp = client.delete(
            f"{self.DATA_ENDPOINT}/sources/{source_id}",
            headers=auth_header(token),
        )
        assert resp.status_code == 200
        assert resp.json()["message"] == "删除成功"

    def test_delete_source_not_found(self, client: TestClient, db: Session):
        """删除不存在的源"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.delete(
            f"{self.DATA_ENDPOINT}/sources/99999",
            headers=auth_header(token),
        )
        assert resp.status_code == 404


class TestDataImport:
    """Excel导入测试"""

    def test_import_excel_success(self, client: TestClient, db: Session):
        """导入Excel数据"""
        create_test_user(db)
        token = get_token_for_user(client)

        # 先创建KPI
        create_test_kpi(db, kpi_code="F_REVENUE_IMP", kpi_name="导入测试KPI")

        # 构造Excel文件
        df = pd.DataFrame({
            "kpi_code": ["F_REVENUE_IMP", "F_REVENUE_IMP"],
            "period": ["2026-06", "2026-07"],
            "actual_value": [100.0, 200.0],
        })
        buffer = io.BytesIO()
        df.to_excel(buffer, index=False)
        buffer.seek(0)

        resp = client.post(
            f"/api/cma/data/import-excel",
            headers=auth_header(token),
            files={"file": ("test.xlsx", buffer, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
        )
        assert resp.status_code == 200
        data = resp.json()
        assert "导入成功" in data["message"]
        assert "2 条" in data["message"]
        assert "batch" in data

    def test_import_excel_missing_columns(self, client: TestClient, db: Session):
        """导入缺少必要列的Excel"""
        create_test_user(db)
        token = get_token_for_user(client)

        df = pd.DataFrame({"name": ["test"], "value": [100]})
        buffer = io.BytesIO()
        df.to_excel(buffer, index=False)
        buffer.seek(0)

        resp = client.post(
            f"/api/cma/data/import-excel",
            headers=auth_header(token),
            files={"file": ("bad.xlsx", buffer, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
        )
        assert resp.status_code == 400

    def test_import_excel_no_match_kpi(self, client: TestClient, db: Session):
        """导入时KPI不存在 → 跳过（不报错）"""
        create_test_user(db)
        token = get_token_for_user(client)

        df = pd.DataFrame({
            "kpi_code": ["NON_EXISTENT_KPI"],
            "period": ["2026-06"],
            "actual_value": [999.0],
        })
        buffer = io.BytesIO()
        df.to_excel(buffer, index=False)
        buffer.seek(0)

        resp = client.post(
            f"/api/cma/data/import-excel",
            headers=auth_header(token),
            files={"file": ("test.xlsx", buffer, "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
        )
        assert resp.status_code == 200
        assert "导入成功" in resp.json()["message"]
        assert "0 条" in resp.json()["message"]

    def test_import_template_download(self, client: TestClient, db: Session):
        """下载导入模板"""
        create_test_user(db)
        token = get_token_for_user(client)

        resp = client.get(
            f"/api/cma/data/import-template",
            headers=auth_header(token),
        )
        # 如果模板文件存在则返回200，否则404（两种情况都算合理）
        assert resp.status_code in (200, 404)