416 lines
13 KiB
Python
416 lines
13 KiB
Python
"""成本计算服务单元测试
|
|
|
|
测试 CostService 的核心业务逻辑
|
|
"""
|
|
|
|
import pytest
|
|
from decimal import Decimal
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.services.cost_service import CostService
|
|
from app.schemas.project_cost import AllocationMethod, CostItemType
|
|
from app.models import (
|
|
Material, Equipment, StaffLevel, Project, FixedCost,
|
|
ProjectCostItem, ProjectLaborCost, ProjectCostSummary
|
|
)
|
|
|
|
|
|
class TestCostService:
|
|
"""成本服务测试类"""
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_get_material_info(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_material: Material
|
|
):
|
|
"""测试获取耗材信息"""
|
|
service = CostService(db_session)
|
|
|
|
# 获取存在的耗材
|
|
material = await service.get_material_info(sample_material.id)
|
|
assert material is not None
|
|
assert material.material_name == "冷凝胶"
|
|
assert material.unit_price == Decimal("2.00")
|
|
|
|
# 获取不存在的耗材
|
|
material = await service.get_material_info(99999)
|
|
assert material is None
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_get_equipment_info(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_equipment: Equipment
|
|
):
|
|
"""测试获取设备信息"""
|
|
service = CostService(db_session)
|
|
|
|
# 获取存在的设备
|
|
equipment = await service.get_equipment_info(sample_equipment.id)
|
|
assert equipment is not None
|
|
assert equipment.equipment_name == "光子仪"
|
|
assert equipment.depreciation_per_use == Decimal("47.50")
|
|
|
|
# 获取不存在的设备
|
|
equipment = await service.get_equipment_info(99999)
|
|
assert equipment is None
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_get_staff_level_info(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_staff_level: StaffLevel
|
|
):
|
|
"""测试获取人员级别信息"""
|
|
service = CostService(db_session)
|
|
|
|
# 获取存在的级别
|
|
level = await service.get_staff_level_info(sample_staff_level.id)
|
|
assert level is not None
|
|
assert level.level_name == "中级美容师"
|
|
assert level.hourly_rate == Decimal("50.00")
|
|
|
|
# 获取不存在的级别
|
|
level = await service.get_staff_level_info(99999)
|
|
assert level is None
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_material_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project_with_costs: Project
|
|
):
|
|
"""测试耗材成本计算"""
|
|
service = CostService(db_session)
|
|
|
|
total, breakdown = await service.calculate_material_cost(
|
|
sample_project_with_costs.id
|
|
)
|
|
|
|
assert total == Decimal("40.00") # 20 * 2.00
|
|
assert len(breakdown) == 1
|
|
assert breakdown[0]["name"] == "冷凝胶"
|
|
assert breakdown[0]["quantity"] == 20.0
|
|
assert breakdown[0]["total"] == 40.0
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_equipment_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project_with_costs: Project
|
|
):
|
|
"""测试设备折旧成本计算"""
|
|
service = CostService(db_session)
|
|
|
|
total, breakdown = await service.calculate_equipment_cost(
|
|
sample_project_with_costs.id
|
|
)
|
|
|
|
assert total == Decimal("47.50") # 1 * 47.50
|
|
assert len(breakdown) == 1
|
|
assert breakdown[0]["name"] == "光子仪"
|
|
assert breakdown[0]["depreciation_per_use"] == 47.5
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_labor_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project_with_costs: Project
|
|
):
|
|
"""测试人工成本计算"""
|
|
service = CostService(db_session)
|
|
|
|
total, breakdown = await service.calculate_labor_cost(
|
|
sample_project_with_costs.id
|
|
)
|
|
|
|
assert total == Decimal("50.00") # 60分钟 / 60 * 50
|
|
assert len(breakdown) == 1
|
|
assert breakdown[0]["name"] == "中级美容师"
|
|
assert breakdown[0]["duration_minutes"] == 60
|
|
assert breakdown[0]["hourly_rate"] == 50.0
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_fixed_cost_allocation_by_count(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_fixed_cost: FixedCost
|
|
):
|
|
"""测试固定成本按项目数量分摊"""
|
|
service = CostService(db_session)
|
|
|
|
allocation, detail = await service.calculate_fixed_cost_allocation(
|
|
sample_project.id,
|
|
method=AllocationMethod.COUNT
|
|
)
|
|
|
|
# 只有一个项目，分摊全部固定成本
|
|
assert allocation == Decimal("30000.00")
|
|
assert detail["method"] == "count"
|
|
assert detail["project_count"] == 1
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_fixed_cost_allocation_by_duration(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_fixed_cost: FixedCost
|
|
):
|
|
"""测试固定成本按时长分摊"""
|
|
service = CostService(db_session)
|
|
|
|
allocation, detail = await service.calculate_fixed_cost_allocation(
|
|
sample_project.id,
|
|
method=AllocationMethod.DURATION
|
|
)
|
|
|
|
# 只有一个项目，占比 100%
|
|
assert allocation == Decimal("30000.00")
|
|
assert detail["method"] == "duration"
|
|
assert detail["project_duration"] == 60
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_project_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project_with_costs: Project,
|
|
sample_fixed_cost: FixedCost
|
|
):
|
|
"""测试项目总成本计算"""
|
|
service = CostService(db_session)
|
|
|
|
result = await service.calculate_project_cost(
|
|
sample_project_with_costs.id,
|
|
allocation_method=AllocationMethod.COUNT
|
|
)
|
|
|
|
assert result.project_id == sample_project_with_costs.id
|
|
assert result.project_name == "光子嫩肤"
|
|
|
|
# 验证成本构成
|
|
breakdown = result.cost_breakdown
|
|
assert breakdown["material_cost"]["subtotal"] == 40.0
|
|
assert breakdown["equipment_cost"]["subtotal"] == 47.5
|
|
assert breakdown["labor_cost"]["subtotal"] == 50.0
|
|
|
|
# 总成本 = 耗材40 + 设备47.5 + 人工50 + 固定30000
|
|
expected_total = 40 + 47.5 + 50 + 30000
|
|
assert result.total_cost == expected_total
|
|
assert result.min_price_suggestion == expected_total
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_calculate_project_cost_not_found(
|
|
self,
|
|
db_session: AsyncSession
|
|
):
|
|
"""测试项目不存在时的错误处理"""
|
|
service = CostService(db_session)
|
|
|
|
with pytest.raises(ValueError, match="项目不存在"):
|
|
await service.calculate_project_cost(99999)
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_add_cost_item_material(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_material: Material
|
|
):
|
|
"""测试添加耗材成本明细"""
|
|
service = CostService(db_session)
|
|
|
|
cost_item = await service.add_cost_item(
|
|
project_id=sample_project.id,
|
|
item_type=CostItemType.MATERIAL,
|
|
item_id=sample_material.id,
|
|
quantity=10,
|
|
remark="测试备注"
|
|
)
|
|
|
|
assert cost_item.project_id == sample_project.id
|
|
assert cost_item.item_type == "material"
|
|
assert cost_item.quantity == Decimal("10")
|
|
assert cost_item.unit_cost == Decimal("2.00")
|
|
assert cost_item.total_cost == Decimal("20.00")
|
|
assert cost_item.remark == "测试备注"
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_add_cost_item_equipment(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_equipment: Equipment
|
|
):
|
|
"""测试添加设备折旧成本明细"""
|
|
service = CostService(db_session)
|
|
|
|
cost_item = await service.add_cost_item(
|
|
project_id=sample_project.id,
|
|
item_type=CostItemType.EQUIPMENT,
|
|
item_id=sample_equipment.id,
|
|
quantity=1,
|
|
)
|
|
|
|
assert cost_item.item_type == "equipment"
|
|
assert cost_item.unit_cost == Decimal("47.50")
|
|
assert cost_item.total_cost == Decimal("47.50")
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_add_cost_item_not_found(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project
|
|
):
|
|
"""测试添加不存在的耗材/设备时的错误处理"""
|
|
service = CostService(db_session)
|
|
|
|
with pytest.raises(ValueError, match="耗材不存在"):
|
|
await service.add_cost_item(
|
|
project_id=sample_project.id,
|
|
item_type=CostItemType.MATERIAL,
|
|
item_id=99999,
|
|
quantity=1,
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="设备不存在"):
|
|
await service.add_cost_item(
|
|
project_id=sample_project.id,
|
|
item_type=CostItemType.EQUIPMENT,
|
|
item_id=99999,
|
|
quantity=1,
|
|
)
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_add_labor_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_staff_level: StaffLevel
|
|
):
|
|
"""测试添加人工成本"""
|
|
service = CostService(db_session)
|
|
|
|
labor_cost = await service.add_labor_cost(
|
|
project_id=sample_project.id,
|
|
staff_level_id=sample_staff_level.id,
|
|
duration_minutes=30,
|
|
remark="测试人工"
|
|
)
|
|
|
|
assert labor_cost.project_id == sample_project.id
|
|
assert labor_cost.duration_minutes == 30
|
|
assert labor_cost.hourly_rate == Decimal("50.00")
|
|
# 30分钟 / 60 * 50 = 25
|
|
assert labor_cost.labor_cost == Decimal("25.00")
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_add_labor_cost_not_found(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project
|
|
):
|
|
"""测试添加不存在的人员级别时的错误处理"""
|
|
service = CostService(db_session)
|
|
|
|
with pytest.raises(ValueError, match="人员级别不存在"):
|
|
await service.add_labor_cost(
|
|
project_id=sample_project.id,
|
|
staff_level_id=99999,
|
|
duration_minutes=30,
|
|
)
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_update_cost_item(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_material: Material
|
|
):
|
|
"""测试更新成本明细"""
|
|
service = CostService(db_session)
|
|
|
|
# 先添加
|
|
cost_item = await service.add_cost_item(
|
|
project_id=sample_project.id,
|
|
item_type=CostItemType.MATERIAL,
|
|
item_id=sample_material.id,
|
|
quantity=10,
|
|
)
|
|
|
|
# 更新数量
|
|
updated = await service.update_cost_item(
|
|
cost_item=cost_item,
|
|
quantity=20,
|
|
remark="更新后备注"
|
|
)
|
|
|
|
assert updated.quantity == Decimal("20")
|
|
assert updated.total_cost == Decimal("40.00") # 20 * 2
|
|
assert updated.remark == "更新后备注"
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_update_labor_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project,
|
|
sample_staff_level: StaffLevel
|
|
):
|
|
"""测试更新人工成本"""
|
|
service = CostService(db_session)
|
|
|
|
# 先添加
|
|
labor = await service.add_labor_cost(
|
|
project_id=sample_project.id,
|
|
staff_level_id=sample_staff_level.id,
|
|
duration_minutes=30,
|
|
)
|
|
|
|
# 更新时长
|
|
updated = await service.update_labor_cost(
|
|
labor_item=labor,
|
|
duration_minutes=60,
|
|
)
|
|
|
|
assert updated.duration_minutes == 60
|
|
assert updated.labor_cost == Decimal("50.00") # 60/60 * 50
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.asyncio
|
|
async def test_empty_project_cost(
|
|
self,
|
|
db_session: AsyncSession,
|
|
sample_project: Project
|
|
):
|
|
"""测试没有成本明细的项目计算"""
|
|
service = CostService(db_session)
|
|
|
|
# 计算空项目成本（无固定成本）
|
|
total_material, _ = await service.calculate_material_cost(sample_project.id)
|
|
total_equipment, _ = await service.calculate_equipment_cost(sample_project.id)
|
|
total_labor, _ = await service.calculate_labor_cost(sample_project.id)
|
|
|
|
assert total_material == Decimal("0")
|
|
assert total_equipment == Decimal("0")
|
|
assert total_labor == Decimal("0")
|