kx_guo 5c7163d79a 统一供应商统计口径并优化后端查询性能
统一对话页与供应商画像的采购订单、异常数、基础分、频次系数、供应商得分和层级计算;补充供应商维度计算结果字段,保留未归类品类标记并优化供应商列表聚合查询;增加响应压缩并完善供应商统计测试。
2026-09-09 10:45:38 +08:00

93 lines
3.6 KiB
Python

from datetime import date
from app.schemas.suppliers import SupplierListData, SupplierOption
from app.services.suppliers import SupplierProfileService
def test_supplier_tier_rules_follow_prd() -> None:
assert SupplierProfileService.tier_for(0, 0, 0) == "新供应商"
assert SupplierProfileService.tier_for(12, 0, 0) == "S级"
assert SupplierProfileService.tier_for(40, 0, 4) == "A级"
assert SupplierProfileService.tier_for(70, 0, 10) == "B级"
assert SupplierProfileService.tier_for(40, 0, 16) == "B级"
assert SupplierProfileService.tier_for(10, 1, 0) == "C级"
assert SupplierProfileService.tier_for(121, 0, 0) == "C级"
def test_supplier_alpha_is_capped_and_rounded() -> None:
assert SupplierProfileService.alpha(0) == 1.0
assert SupplierProfileService.alpha(42) == 1.13
assert SupplierProfileService.alpha(148) == 1.44
assert SupplierProfileService.alpha(200) == 1.5
def test_supplier_list_exposes_default_highest_anomaly_supplier() -> None:
result = SupplierListData(
items=[
SupplierOption(name="甲供应商", anomaly_count=2, tier="S级"),
SupplierOption(name="乙供应商", anomaly_count=1, tier="S级"),
],
default_supplier="甲供应商",
)
assert result.default_supplier == result.items[0].name
def test_supplier_category_breakdown_is_top_five_with_material_and_problem() -> None:
rows = [
{"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
{"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
{"product_category": "面霜", "product_name": "面霜", "material_type": "瓶子", "anomaly_type": "刮花"},
]
result = SupplierProfileService._categories(rows)
assert result[0].category == "洗发水/洗护类"
assert result[0].top_material == "泵头"
assert result[0].top_problem == "漏液"
def test_unclassified_category_is_marked_and_sorted_after_real_categories() -> None:
rows = [
{"product_category": None, "product_name": "未命名产品", "material_type": "瓶子", "anomaly_type": "刮花"},
{"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
]
result = SupplierProfileService._categories(rows)
assert result[0].category == "洗发水/洗护类"
assert result[0].is_unclassified is False
assert result[1].category == "待补标品类"
assert result[1].is_unclassified is True
def test_display_categories_keeps_unclassified_after_five_real_categories() -> None:
rows = [
{"product_category": category, "product_name": category, "material_type": "瓶子", "anomaly_type": "异常"}
for category in ["洗发水", "面霜", "沐浴露", "牙膏", "清洁剂", "脱毛膏"]
] + [{"product_category": None, "product_name": "未命名", "material_type": "瓶子", "anomaly_type": "异常"}]
result = SupplierProfileService._display_categories(SupplierProfileService._categories(rows))
assert len(result) == 6
assert result[-1].is_unclassified is True
def test_supplier_recent_records_are_limited_and_sorted() -> None:
service = SupplierProfileService(object(), as_of=date(2026, 9, 7))
rows = [
{
"id": index,
"date": date(2026, 4, index),
"product_name": "产品",
"anomaly_type": str(index),
"defect_level": "B",
}
for index in range(1, 13)
]
result = service._recent_records(rows)
assert len(result) == 10
assert result[0].case_id == 12