205 lines
6.8 KiB
Python
205 lines
6.8 KiB
Python
"""成本分析模块测试 — 标准成本/实际成本CRUD + ABC + 看板"""
|
|
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
|
|
from app.models import StandardCost, ActualCost, AbcActivity
|
|
|
|
|
|
class TestStandardCost:
|
|
"""标准成本卡片CRUD测试"""
|
|
|
|
BASE = "/api/cma/cost"
|
|
|
|
def test_list_standard_costs_empty(self, client: TestClient, db: Session):
|
|
"""空列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.get(f"{self.BASE}/standard-costs", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert resp.json()["data"] == []
|
|
|
|
def test_create_standard_cost(self, client: TestClient, db: Session):
|
|
"""创建标准成本卡片"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
f"{self.BASE}/standard-costs",
|
|
headers=auth_header(token),
|
|
json={
|
|
"product_code": "PROD_A",
|
|
"product_name": "产品A",
|
|
"cost_type": "material",
|
|
"item_name": "原材料X",
|
|
"standard_quantity": 10.0,
|
|
"standard_price": 5.0,
|
|
"unit": "kg",
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "标准成本已创建"
|
|
|
|
# 验证列表中有数据
|
|
list_resp = client.get(f"{self.BASE}/standard-costs", headers=auth_header(token))
|
|
assert len(list_resp.json()["data"]) == 1
|
|
|
|
def test_update_standard_cost(self, client: TestClient, db: Session):
|
|
"""修改标准成本卡片"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
sc = StandardCost(
|
|
product_code="PROD_B", product_name="产品B", cost_type="labor",
|
|
item_name="人工", standard_quantity=100, unit="小时",
|
|
standard_price=50.0, standard_cost=5000.0,
|
|
)
|
|
db.add(sc)
|
|
db.commit()
|
|
|
|
resp = client.put(
|
|
f"{self.BASE}/standard-costs/{sc.id}",
|
|
headers=auth_header(token),
|
|
json={"standard_price": 55.0, "remark": "调薪后更新"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已更新"
|
|
|
|
def test_delete_standard_cost(self, client: TestClient, db: Session):
|
|
"""删除（归档）标准成本卡片"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
sc = StandardCost(
|
|
product_code="PROD_C", product_name="产品C", cost_type="overhead",
|
|
item_name="水电费", standard_quantity=1, unit="项",
|
|
standard_price=1000.0, standard_cost=1000.0,
|
|
)
|
|
db.add(sc)
|
|
db.commit()
|
|
|
|
resp = client.delete(
|
|
f"{self.BASE}/standard-costs/{sc.id}",
|
|
headers=auth_header(token),
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "已归档"
|
|
|
|
|
|
class TestActualCost:
|
|
"""实际成本CRUD测试"""
|
|
|
|
BASE = "/api/cma/cost"
|
|
|
|
def test_create_actual_cost(self, client: TestClient, db: Session):
|
|
"""录入实际成本"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
f"{self.BASE}/actual-costs",
|
|
headers=auth_header(token),
|
|
json={
|
|
"period": "2026-06",
|
|
"product_code": "PROD_A",
|
|
"product_name": "产品A",
|
|
"cost_type": "material",
|
|
"item_name": "原材料X",
|
|
"actual_quantity": 12.0,
|
|
"actual_price": 4.8,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "实际成本已录入"
|
|
|
|
def test_list_actual_costs(self, client: TestClient, db: Session):
|
|
"""查询实际成本"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
ac = ActualCost(
|
|
period="2026-06", product_code="PROD_A", product_name="产品A",
|
|
cost_type="material", item_name="原材料X",
|
|
actual_quantity=10, actual_price=5.0, actual_cost=50.0,
|
|
)
|
|
db.add(ac)
|
|
db.commit()
|
|
|
|
resp = client.get(f"{self.BASE}/actual-costs?period=2026-06", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["data"]) == 1
|
|
|
|
|
|
class TestAbcCost:
|
|
"""ABC作业成本测试"""
|
|
|
|
BASE = "/api/cma/cost"
|
|
|
|
def test_create_abc_activity(self, client: TestClient, db: Session):
|
|
"""创建ABC作业中心"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
resp = client.post(
|
|
f"{self.BASE}/abc/activities",
|
|
headers=auth_header(token),
|
|
json={
|
|
"activity_code": "ACT_001",
|
|
"activity_name": "机器调试",
|
|
"cost_driver": "调试次数",
|
|
"driver_unit": "次",
|
|
"total_cost": 50000.0,
|
|
"driver_volume": 100,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "作业中心已创建"
|
|
|
|
def test_list_abc_activities(self, client: TestClient, db: Session):
|
|
"""查询ABC作业中心列表"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
act = AbcActivity(
|
|
activity_code="ACT_002", activity_name="品质检验",
|
|
cost_driver="检验批次", driver_unit="批",
|
|
total_cost=30000.0, driver_volume=200, driver_rate=150.0,
|
|
)
|
|
db.add(act)
|
|
db.commit()
|
|
|
|
resp = client.get(f"{self.BASE}/abc/activities", headers=auth_header(token))
|
|
assert resp.status_code == 200
|
|
assert len(resp.json()["data"]) >= 1
|
|
assert resp.json()["data"][0]["activity_code"] == "ACT_002"
|
|
|
|
def test_abc_allocate(self, client: TestClient, db: Session):
|
|
"""执行ABC成本分配"""
|
|
create_test_user(db)
|
|
token = get_token_for_user(client)
|
|
|
|
act = AbcActivity(
|
|
activity_code="ACT_003", activity_name="订单处理",
|
|
cost_driver="订单数", driver_unit="单",
|
|
total_cost=20000.0, driver_volume=400, driver_rate=50.0,
|
|
)
|
|
db.add(act)
|
|
db.commit()
|
|
|
|
resp = client.post(
|
|
f"{self.BASE}/abc/allocate",
|
|
headers=auth_header(token),
|
|
json={
|
|
"activity_id": act.id,
|
|
"period": "2026-06",
|
|
"product_code": "PROD_A",
|
|
"product_name": "产品A",
|
|
"driver_consumed": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# 30 * 50 = 1500
|
|
assert data["allocated_cost"] == 1500.0
|