feat(cost): 成本模型加研发归集维度 research_flag/project_type (cma-rd-dimension-20260905)

- StandardCost/ActualCost 增加 research_flag(int,idx)/project_type(varchar30,idx)
  project_type: platform_dev平台研发/client_delivery客户交付/internal内部管理
- 生产库两表 ALTER 完成, 存量 29+134 行默认 internal(保守防误归研发)
- API: create/update 支持写入维度, 未传默认 client_delivery(录入默认),
  research_flag 未显式传入按 project_type 联动(platform_dev=1);
  create/update/delete 绑定 entity_id(token优先, 防跨账套)
- API: standard/actual 列表支持 project_type/research_flag 过滤
- 前端 CostDashboard: 表单加归集类型下拉(默认客户交付+研发提示),
  列表加归集类型列(标签), 工具栏按归集类型过滤
- tests: test_cost.py +6 用例(默认值/联动/过滤/更新)
- scripts: verify_cost_research_dimension_e2e.py 生产E2E(标记+清理+回查)
This commit is contained in:
Hermes CI Fix
2026-09-05 09:10:35 +08:00
parent 46096e8f84
commit 696f8bc934
5 changed files with 451 additions and 8 deletions
+146
View File
@@ -45,6 +45,99 @@ class TestStandardCost:
list_resp = client.get(f"{self.BASE}/standard-costs", headers=auth_header(token))
assert len(list_resp.json()["data"]) == 1
def test_create_standard_cost_default_client_delivery(self, client: TestClient, db: Session):
"""未传归集维度 → 默认 client_delivery/research_flag=0(录入默认,客户交付)"""
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_DEF",
"product_name": "默认产品",
"cost_type": "labor",
"item_name": "人工",
"standard_quantity": 1.0,
"standard_price": 100.0,
},
)
assert resp.status_code == 200
row = db.query(StandardCost).filter(StandardCost.product_code == "PROD_DEF").first()
assert row is not None
assert row.project_type == "client_delivery"
assert row.research_flag == 0
def test_create_standard_cost_platform_dev_research_flag_linked(self, client: TestClient, db: Session):
"""project_type=platform_dev → research_flag 自动联动为 1(研发归集)"""
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_RD",
"product_name": "CMA平台开发",
"cost_type": "labor",
"item_name": "研发工时",
"standard_quantity": 10.0,
"standard_price": 80.0,
"project_type": "platform_dev",
},
)
assert resp.status_code == 200
row = db.query(StandardCost).filter(StandardCost.product_code == "PROD_RD").first()
assert row.project_type == "platform_dev"
assert row.research_flag == 1
def test_list_standard_costs_filter_by_project_type(self, client: TestClient, db: Session):
"""列表按 project_type 过滤:研发只返回平台研发,不混入客户交付"""
create_test_user(db)
token = get_token_for_user(client)
for code, pt in (("RD_X", "platform_dev"), ("CD_X", "client_delivery"), ("IM_X", "internal")):
db.add(StandardCost(
product_code=code, product_name=code, cost_type="overhead",
item_name=code, standard_quantity=1, standard_price=1.0,
standard_cost=1.0,
project_type=pt, research_flag=1 if pt == "platform_dev" else 0,
))
db.commit()
resp = client.get(f"{self.BASE}/standard-costs?project_type=platform_dev", headers=auth_header(token))
assert resp.status_code == 200
codes = [i["product_code"] for i in resp.json()["data"]]
assert codes == ["RD_X"]
resp2 = client.get(f"{self.BASE}/standard-costs?research_flag=1", headers=auth_header(token))
assert resp2.status_code == 200
codes2 = [i["product_code"] for i in resp2.json()["data"]]
assert codes2 == ["RD_X"]
def test_update_standard_cost_project_type(self, client: TestClient, db: Session):
"""更新归集类型 → research_flag 联动(client_delivery 改为 platform_dev)"""
create_test_user(db)
token = get_token_for_user(client)
sc = StandardCost(
product_code="PROD_UPD", product_name="产品U", cost_type="labor",
item_name="人工", standard_quantity=100, unit="小时",
standard_price=50.0, standard_cost=5000.0,
project_type="client_delivery", research_flag=0,
)
db.add(sc)
db.commit()
resp = client.put(
f"{self.BASE}/standard-costs/{sc.id}",
headers=auth_header(token),
json={"project_type": "platform_dev"},
)
assert resp.status_code == 200
db.refresh(sc)
assert sc.project_type == "platform_dev"
assert sc.research_flag == 1
def test_update_standard_cost(self, client: TestClient, db: Session):
"""修改标准成本卡片"""
create_test_user(db)
@@ -130,6 +223,59 @@ class TestActualCost:
assert resp.status_code == 200
assert len(resp.json()["data"]) == 1
def test_create_actual_cost_research_dims(self, client: TestClient, db: Session):
"""录入实际成本带归集维度(平台研发 → research_flag=1)"""
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-09",
"product_code": "RD_SVC",
"product_name": "CMA平台化开发",
"cost_type": "labor",
"item_name": "平台研发工时",
"actual_quantity": 40.0,
"actual_price": 100.0,
"project_type": "platform_dev",
},
)
assert resp.status_code == 200
row = db.query(ActualCost).filter(ActualCost.product_code == "RD_SVC").first()
assert row is not None
assert row.entity_id == 1
assert row.project_type == "platform_dev"
assert row.research_flag == 1
def test_list_actual_costs_filter_by_research(self, client: TestClient, db: Session):
"""实际成本列表按归集类型/研发标记过滤"""
create_test_user(db)
token = get_token_for_user(client)
db.add(ActualCost(
period="2026-09", product_code="RD_SVC", product_name="平台开发",
cost_type="labor", item_name="研发工时", actual_quantity=1, actual_price=1.0,
actual_cost=1.0,
project_type="platform_dev", research_flag=1,
))
db.add(ActualCost(
period="2026-09", product_code="CD_SVC", product_name="客户交付",
cost_type="labor", item_name="实施工时", actual_quantity=1, actual_price=1.0,
actual_cost=1.0,
project_type="client_delivery", research_flag=0,
))
db.commit()
resp = client.get(f"{self.BASE}/actual-costs?project_type=platform_dev&period=2026-09", headers=auth_header(token))
assert resp.status_code == 200
codes = [i["product_code"] for i in resp.json()["data"]]
assert codes == ["RD_SVC"]
resp2 = client.get(f"{self.BASE}/actual-costs?research_flag=0&period=2026-09", headers=auth_header(token))
codes2 = [i["product_code"] for i in resp2.json()["data"]]
assert codes2 == ["CD_SVC"]
class TestAbcCost:
"""ABC作业成本测试"""