group_fqcd_jr/tests/unit/service/test_promotion_material.py

from pathlib import Path

import pytest

from app.service.promotion_compliance import PromotionComplianceChecker
from app.service.promotion_performance import (
    create_performance_chart,
    parse_performance_file,
)


def _valid_inputs() -> dict[str, object]:
    return {
        "product_info": {
            "fund_type": "混合型",
            "operation_mode": "开放式",
            "investment_objective": "长期配置",
        },
        "manager_info": {
            "manager_name": "张三",
            "management_company": "奶龙基金管理有限公司",
            "registration_code": "P10000001",
        },
        "team_info": {"team_description": "具备完整投研分工"},
        "strategy_info": {
            "investment_scope": "股票和债券",
            "strategy": "自上而下配置",
            "restrictions": "遵守法律法规和基金合同",
        },
        "fee_structure": {
            "subscription_fee": "1.0%",
            "purchase_fee": "1.0%",
            "redemption_fee": "0.5%",
            "sales_service_fee": "0.2%",
            "management_fee": "1.0%",
            "custody_fee": "0.2%",
            "client_maintenance_fee": "不适用",
        },
        "performance_info": {
            "show_product_performance": True,
            "history_months": 12,
            "performance_attachment_id": 1,
            "product_return": "8%",
            "max_drawdown": "-5%",
            "volatility": "10%",
            "sharpe_ratio": "0.8",
        },
        "risk_disclosure": {},
    }


def test_performance_requires_more_than_six_months_and_risk_metrics() -> None:
    checker = PromotionComplianceChecker()
    inputs = _valid_inputs()
    inputs["performance_info"] = {
        "show_product_performance": True,
        "history_months": 6,
        "product_return": "8%",
    }

    findings = checker.check_inputs(inputs)

    codes = {finding.rule_code for finding in findings}
    assert "performance.history_short" in codes

    draft_findings = checker.check_draft({"text": "展示产品业绩"}, inputs)
    assert "performance.risk_metrics_missing" in {
        finding.rule_code for finding in draft_findings
    }


def test_performance_ranking_requires_public_three_year_source() -> None:
    checker = PromotionComplianceChecker()
    inputs = _valid_inputs()
    inputs["performance_info"] = {
        **inputs["performance_info"],
        "ranking": {
            "enabled": True,
            "institution_name": "评价机构",
            "evaluation_period_years": "2年",
            "ranking_text": "排名靠前",
        },
    }

    findings = checker.check_inputs(inputs)

    assert any(
        finding.rule_code == "performance.ranking_source_invalid"
        for finding in findings
    )


def test_fee_structure_requires_every_disclosed_item() -> None:
    checker = PromotionComplianceChecker()
    inputs = _valid_inputs()
    inputs["fee_structure"] = {"subscription_fee": "1.0%"}

    findings = checker.check_inputs(inputs)

    assert any(finding.rule_code == "fee_structure.incomplete" for finding in findings)


def test_performance_display_requires_data_attachment() -> None:
    checker = PromotionComplianceChecker()
    inputs = _valid_inputs()
    inputs["performance_info"] = {
        **inputs["performance_info"],
        "performance_attachment_id": None,
    }

    findings = checker.check_inputs(inputs)

    assert any(
        finding.rule_code == "performance.data_attachment_missing"
        for finding in findings
    )


def test_performance_file_is_validated_and_chart_is_created(tmp_path: Path) -> None:
    source = tmp_path / "performance.csv"
    source.write_text(
        "日期,产品收益率,业绩比较基准收益率,基金经理代表产品收益率\n"
        "2025-01-01,1%,0.5%,0.8%\n"
        "2025-02-01,2%,1%,1.6%\n",
        encoding="utf-8-sig",
    )

    data = parse_performance_file(source)
    output = tmp_path / "chart.png"

    create_performance_chart(data, output, title="测试曲线")

    assert data["row_count"] == 2
    assert output.exists()
    assert output.stat().st_size > 0


def test_performance_chart_contains_chinese_text_without_missing_font_failure(
    tmp_path: Path,
) -> None:
    source = tmp_path / "performance.csv"
    source.write_text(
        "日期,产品收益率,业绩比较基准收益率\n"
        "2025-01-01,1%,0.5%\n"
        "2025-02-01,2%,1%\n",
        encoding="utf-8-sig",
    )

    output = tmp_path / "chart-with-chinese.png"
    data = parse_performance_file(source)
    create_performance_chart(data, output, title="产品业绩曲线")

    assert output.is_file()
    assert output.stat().st_size > 0


def test_performance_file_rejects_duplicate_dates(tmp_path: Path) -> None:
    source = tmp_path / "duplicate.csv"
    source.write_text(
        "日期,产品收益率\n2025-01-01,1%\n2025-01-01,2%\n",
        encoding="utf-8-sig",
    )

    with pytest.raises(ValueError, match="重复日期"):
        parse_performance_file(source)


def test_performance_xlsx_cells_are_json_safe_and_file_handle_is_released(
    tmp_path: Path,
) -> None:
    """回归上传业绩 Excel 报 500 的问题。

    openpyxl 把日期单元格读成 datetime，直接写 MySQL JSON 列会抛
    "Object of type datetime is not JSON serializable"；同时 read_only 工作簿
    不关闭会一直占用上传文件。这里同时守住这两个行为。
    """
    import json
    from datetime import date

    from openpyxl import Workbook

    source = tmp_path / "performance.xlsx"
    book = Workbook()
    sheet = book.active
    sheet.append(["日期", "产品收益率", "业绩比较基准收益率"])
    sheet.append([date(2025, 1, 31), 1.0, 0.5])
    sheet.append([date(2025, 2, 28), 2.0, 1.0])
    book.save(source)
    book.close()

    data = parse_performance_file(source)

    assert data["row_count"] == 2
    assert data["rows"][0]["日期"] == "2025-01-31"
    json.dumps(data)  # 不抛异常才说明可以写入 JSON 列
    source.unlink()  # 不抛异常才说明文件句柄已释放