张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
234 lines
7.2 KiB
Python
234 lines
7.2 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.core.errors import ValidationAgentError
|
|
from app.service.promotion_compliance import PromotionComplianceChecker
|
|
from app.service.promotion_material_service import PromotionMaterialService
|
|
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
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize("attachment_type", ["source_evidence", "template_file"])
|
|
def test_removed_attachment_types_are_rejected(attachment_type: str) -> None:
|
|
with pytest.raises(ValidationAgentError, match="不支持的附件类型"):
|
|
PromotionMaterialService._validate_attachment(
|
|
attachment_type,
|
|
"removed.pdf",
|
|
"application/pdf",
|
|
1024,
|
|
max_photo_size=10 * 1024 * 1024,
|
|
max_performance_size=20 * 1024 * 1024,
|
|
)
|
|
|
|
|
|
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_summary_uses_latest_file_date_and_product_series(tmp_path: Path) -> None:
|
|
source = tmp_path / "summary.csv"
|
|
source.write_text(
|
|
"日期,产品收益率,业绩比较基准收益率\n"
|
|
"2025-01-31,1%,0.5%\n"
|
|
"2025-02-28,5%,1%\n"
|
|
"2025-03-31,3%,1.2%\n"
|
|
"2026-03-31,8%,2%\n",
|
|
encoding="utf-8-sig",
|
|
)
|
|
|
|
data = parse_performance_file(source)
|
|
|
|
assert data["summary"] == {
|
|
"initial_date": "2025-01-31",
|
|
"as_of_date": "2026-03-31",
|
|
"history_months": 14,
|
|
"product_return": "8%",
|
|
"max_drawdown": "-1.9%",
|
|
}
|
|
|
|
|
|
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() # 不抛异常才说明文件句柄已释放
|