135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
"""采购需求推导集成测试
|
||
|
||
覆盖 P4-4 新功能：POST /api/purchase-demands/calculate
|
||
- 正常推导（有 BOM、有库存、有供应商）
|
||
- 库存不足时的缺口计算
|
||
- 无效销售订单 → 404
|
||
- 成品无 BOM → 空结果
|
||
- 空 sales_order_ids → 422
|
||
"""
|
||
import pytest
|
||
|
||
|
||
async def _get_inventory_id(client, sku: str) -> int:
|
||
"""从分页 API 获取库存记录 ID"""
|
||
resp = await client.get("/api/inventory")
|
||
items = resp.json()["items"]
|
||
return next(r["id"] for r in items if r["product_sku"] == sku)
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_purchase_demand_basic_no_shortage(client):
|
||
"""库存充足时：需求量正确计算，缺口为 0"""
|
||
# 创建销售订单：1 套标准模具（自动发料 3 单位，库存从 1000 → 997）
|
||
so_resp = await client.post("/api/sales-orders", json={
|
||
"customer_id": 1,
|
||
"items": [{"product_id": 2, "quantity": 1, "unit_price": 1000.0}],
|
||
})
|
||
assert so_resp.status_code == 201, so_resp.text
|
||
so_id = so_resp.json()["id"]
|
||
|
||
# 重置库存为 1000（覆盖发料消耗）
|
||
inv_id = await _get_inventory_id(client, "MAT-001")
|
||
await client.put(f"/api/inventory/{inv_id}", json={
|
||
"quantity": 1000, "locked_quantity": 0,
|
||
})
|
||
|
||
# 调用采购需求推导
|
||
resp = await client.post("/api/purchase-demands/calculate", json={
|
||
"sales_order_ids": [so_id],
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
|
||
assert data["source_order_ids"] == [so_id]
|
||
assert len(data["source_order_nos"]) == 1
|
||
assert len(data["items"]) == 1
|
||
|
||
item = data["items"][0]
|
||
assert item["material_sku"] == "MAT-001"
|
||
# BOM: qty=2, loss_rate=0.05 → ceil(1*2*1.05) = ceil(2.1) = 3
|
||
# Decimal 在 JSON 中可能序列化为字符串，用 int() 转换
|
||
assert int(item["required_quantity"]) == 3
|
||
# 库存 1000 > 需求 3，无缺口
|
||
assert int(item["shortage_quantity"]) == 0
|
||
assert float(item["estimated_cost"]) == 0
|
||
# 推荐供应商（seeded_db 中 MaterialSupplier is_primary=True, lead_time=7）
|
||
assert item["suggested_supplier_name"] == "供应商A"
|
||
assert item["supplier_lead_time"] == 7
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_purchase_demand_with_shortage(client):
|
||
"""库存不足时：缺口 = 需求 - 库存，预计金额 = 缺口 × 单价"""
|
||
# 创建销售订单：1 套标准模具（自动发料 3 单位，库存从 1000 → 997）
|
||
so_resp = await client.post("/api/sales-orders", json={
|
||
"customer_id": 1,
|
||
"items": [{"product_id": 2, "quantity": 1, "unit_price": 1000.0}],
|
||
})
|
||
assert so_resp.status_code == 201, so_resp.text
|
||
so_id = so_resp.json()["id"]
|
||
|
||
# 将库存设为 0（完全缺货）
|
||
inv_id = await _get_inventory_id(client, "MAT-001")
|
||
await client.put(f"/api/inventory/{inv_id}", json={
|
||
"quantity": 0, "locked_quantity": 0,
|
||
})
|
||
|
||
# 调用采购需求推导
|
||
resp = await client.post("/api/purchase-demands/calculate", json={
|
||
"sales_order_ids": [so_id],
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
|
||
item = data["items"][0]
|
||
# BOM: qty=2, loss_rate=0.05 → ceil(1*2*1.05) = ceil(2.1) = 3
|
||
assert int(item["required_quantity"]) == 3
|
||
# 库存 0
|
||
assert int(item["available_quantity"]) == 0
|
||
# 缺口 = 3 - 0 = 3
|
||
assert int(item["shortage_quantity"]) == 3
|
||
# 预计金额 = 3 × 10.0 = 30.0
|
||
assert float(item["estimated_cost"]) == 30.0
|
||
assert float(data["total_estimated_cost"]) == 30.0
|
||
assert data["shortage_count"] == 1
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_purchase_demand_invalid_order_ids(client):
|
||
"""不存在的销售订单 ID → 404"""
|
||
resp = await client.post("/api/purchase-demands/calculate", json={
|
||
"sales_order_ids": [99999],
|
||
})
|
||
assert resp.status_code == 404
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_purchase_demand_no_bom_returns_empty(client):
|
||
"""成品无 BOM 关联时，推导结果为空"""
|
||
# product_id=3 是无 BOM 的成品（MOLD-NB）
|
||
so_resp = await client.post("/api/sales-orders", json={
|
||
"customer_id": 1,
|
||
"items": [{"product_id": 3, "quantity": 1, "unit_price": 500.0}],
|
||
})
|
||
assert so_resp.status_code == 201, so_resp.text
|
||
so_id = so_resp.json()["id"]
|
||
|
||
resp = await client.post("/api/purchase-demands/calculate", json={
|
||
"sales_order_ids": [so_id],
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["items"] == []
|
||
assert float(data["total_estimated_cost"]) == 0
|
||
assert data["shortage_count"] == 0
|
||
|
||
|
||
@pytest.mark.anyio
|
||
async def test_purchase_demand_empty_request_validation(client):
|
||
"""空 sales_order_ids 列表 → 422 schema 校验失败"""
|
||
resp = await client.post("/api/purchase-demands/calculate", json={
|
||
"sales_order_ids": [],
|
||
})
|
||
assert resp.status_code == 422
|