cjw
64dc85bd14
- inventory业务层下沉（薄路由+service orchestration模式）： - customer/supplier/warehouse -> master_data_service - material_routes（价格历史/趋势/供应商关联）-> material_service - product_routes（CRUD/BOM/from-task跨模块桥接）-> product_service - dashboard_routes（首页统计/低库存预警）-> dashboard_service - inventory侧新增service回归覆盖（dashboard 2 / master_data 10 / material 10 / product 14），含跨模块桥接测试种子 - Pydantic v2弃用清零：全仓14处 class Config 全部迁移到 model_config = ConfigDict(from_attributes=True)（含 shared auth） - datetime.utcnow() 弃用清零：auth_service 3处统一改 datetime.now(timezone.utc) - 同步文档：STATUS / ROADMAP / TECH_DEBT（D12清偿）/ AGENTS 代码地图 测试基线：126 passed, 4 skipped（无deprecation warning） Co-Authored-By: Claude Code <noreply@anthropic.com>
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
import pytest
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_material_price_history_create_updates_latest_cost(client):
|
|
resp = await client.post(
|
|
"/api/materials/1/price-history",
|
|
json={"price": 12.5, "supplier_id": 1, "remark": "latest"},
|
|
)
|
|
assert resp.status_code == 201
|
|
body = resp.json()
|
|
assert body["product_id"] == 1
|
|
assert body["price"] == 12.5
|
|
assert body["supplier_name"] == "供应商A"
|
|
|
|
history = await client.get("/api/materials/1/price-history")
|
|
assert history.status_code == 200
|
|
assert history.json()[0]["price"] == 12.5
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_material_price_trend_returns_change_summary(client):
|
|
resp1 = await client.post(
|
|
"/api/materials/1/price-history",
|
|
json={"price": 10.0, "supplier_id": 1},
|
|
)
|
|
assert resp1.status_code == 201
|
|
resp2 = await client.post(
|
|
"/api/materials/1/price-history",
|
|
json={"price": 15.0, "supplier_id": 1},
|
|
)
|
|
assert resp2.status_code == 201
|
|
|
|
trend = await client.get("/api/materials/1/price-trend", params={"months": 6})
|
|
assert trend.status_code == 200
|
|
body = trend.json()
|
|
prices = [item["price"] for item in body["price_history"]]
|
|
assert sorted(prices) == [10.0, 15.0]
|
|
assert body["current_price"] in {10.0, 15.0}
|
|
assert body["price_change"] in {5.0, -5.0}
|
|
assert body["price_change_percent"] in {50.0, -33.33}
|
|
assert len(body["price_history"]) == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_material_price_trend_requires_history(client):
|
|
resp = await client.get("/api/materials/1/price-trend")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_add_material_supplier_and_list_by_material(client):
|
|
resp = await client.post(
|
|
"/api/materials/1/suppliers",
|
|
json={"supplier_id": 1, "is_primary": True, "lead_time": 7, "min_order_quantity": 10},
|
|
)
|
|
assert resp.status_code == 400 # seeded_db 已存在主关联
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_get_material_suppliers_and_supplier_materials(client):
|
|
by_material = await client.get("/api/materials/1/suppliers")
|
|
assert by_material.status_code == 200
|
|
material_rows = by_material.json()
|
|
assert len(material_rows) == 1
|
|
assert material_rows[0]["supplier_id"] == 1
|
|
assert material_rows[0]["supplier_name"] == "供应商A"
|
|
|
|
by_supplier = await client.get("/api/materials/suppliers/1/materials")
|
|
assert by_supplier.status_code == 200
|
|
supplier_rows = by_supplier.json()
|
|
assert len(supplier_rows) == 1
|
|
assert supplier_rows[0]["product_id"] == 1
|
|
assert supplier_rows[0]["product_sku"] == "MAT-001"
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_remove_material_supplier_deletes_association(client):
|
|
resp = await client.delete("/api/materials/suppliers/1")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["message"] == "物料供应商关联已删除"
|
|
|
|
by_material = await client.get("/api/materials/1/suppliers")
|
|
assert by_material.status_code == 200
|
|
assert by_material.json() == []
|
|
|
|
|
|
@pytest.mark.anyio
|
|
@pytest.mark.parametrize(
|
|
"url,payload,expected",
|
|
[
|
|
("/api/materials/2/price-history", {"price": 10.0}, 400),
|
|
("/api/materials/99999/price-history", {"price": 10.0}, 404),
|
|
("/api/materials/1/price-history", {"price": 10.0, "supplier_id": 99999}, 400),
|
|
("/api/materials/1/suppliers", {"supplier_id": 99999}, 400),
|
|
],
|
|
)
|
|
async def test_material_endpoints_validation(url, payload, expected, client):
|
|
resp = await client.post(url, json=payload)
|
|
assert resp.status_code == expected
|