tangxh/ForgeFlow-ERP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

完善系统拓展与原材料生命周期视图

Browse Source
This commit is contained in:
汤学会
2026-06-14 20:38:53 +08:00
parent 453b19399b
commit df9a4b485d
14 changed files with 5960 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_dashboard_material_lot_lifecycle.py
+510
View File
@@ -0,0 +1,510 @@
from __future__ import annotations
import unittest
from datetime import datetime
from decimal import Decimal
from sqlalchemy import BigInteger, create_engine
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.orm import Session, sessionmaker
@compiles(BigInteger, "sqlite")
def _compile_bigint_for_sqlite(type_, compiler, **kw) -> str:
_ = type_, compiler, kw
return "INTEGER"
import app.models.master_data # noqa: E402,F401
import app.models.miniapp # noqa: E402,F401
import app.models.operations # noqa: E402,F401
import app.models.org # noqa: E402,F401
import app.models.planning # noqa: E402,F401
import app.models.sales # noqa: E402,F401
from app.api.routes.dashboard import get_material_lot_lifecycle, list_dashboard_material_lots # noqa: E402
from app.models.base import Base # noqa: E402
from app.models.master_data import Item, Warehouse # noqa: E402
from app.models.operations import Delivery, DeliveryItem, OperationReport, ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot # noqa: E402
from app.models.org import Department, Employee, SystemConfig # noqa: E402
from app.schemas.dashboard import DashboardMaterialLotRead # noqa: E402
from app.services.material_lifecycle_dag import build_material_lot_lifecycle_graph # noqa: E402
class DashboardMaterialLotLifecycleTest(unittest.TestCase):
def setUp(self) -> None:
engine = create_engine("sqlite+pysqlite:///:memory:", future=True)
Base.metadata.create_all(engine)
self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True)
self.db: Session = self.SessionLocal()
self.now = datetime(2026, 6, 14, 9, 0)
self._seed_lot_and_ledger()
def tearDown(self) -> None:
self.db.close()
def _seed_lot_and_ledger(self) -> None:
raw_item = Item(
id=1,
item_code="RM-001",
item_name="冷轧钢板",
item_type="RAW_MATERIAL",
unit_weight_kg=Decimal("1"),
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
product = Item(
id=2,
item_code="FG-001",
item_name="钢制碗",
item_type="FINISHED_GOOD",
unit_weight_kg=Decimal("0.5"),
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
warehouse = Warehouse(
id=1,
warehouse_code="WH-RAW",
warehouse_name="原材料库",
warehouse_type="RAW",
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
lot = StockLot(
id=1,
lot_no="YL0001",
lot_role="RAW_MATERIAL",
item_id=1,
warehouse_id=1,
source_doc_type="期初入库",
source_doc_id=1,
inbound_qty=Decimal("0"),
inbound_weight_kg=Decimal("500"),
remaining_qty=Decimal("0"),
remaining_weight_kg=Decimal("300"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("4"),
quality_status="PASS",
status="AVAILABLE",
created_at=self.now,
updated_at=self.now,
)
ledger = ProductionBatchLedger(
id=1,
material_lot_id=1,
material_lot_no="YL0001",
material_item_id=1,
product_item_id=2,
status="在生产",
miniapp_selectable=1,
total_issued_weight_kg=Decimal("120"),
outside_weight_kg=Decimal("42"),
finished_inbound_qty=Decimal("100"),
surplus_inbound_weight_kg=Decimal("18"),
scrap_inbound_weight_kg=Decimal("6"),
writeoff_weight_kg=Decimal("54"),
first_issue_time=self.now,
last_issue_time=self.now,
created_at=self.now,
updated_at=self.now,
)
dept = Department(
id=1,
dept_code="D-001",
dept_name="生产部",
dept_type="PRODUCTION",
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
employee = Employee(
id=1,
employee_code="EMP-001",
employee_name="张三",
dept_id=1,
is_operator=1,
is_workshop_staff=1,
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
txns = [
ProductionBatchLedgerTxn(
id=1,
production_ledger_id=1,
txn_type="生产出库",
qty_delta=Decimal("0"),
weight_delta_kg=Decimal("120"),
outside_weight_after_kg=Decimal("120"),
source_doc_type="库存流水",
source_doc_id=101,
biz_time=self.now,
operator_user_id=1,
remark="生产出库累计到生产台账",
created_at=self.now,
updated_at=self.now,
),
ProductionBatchLedgerTxn(
id=2,
production_ledger_id=1,
txn_type="成品入库",
qty_delta=Decimal("100"),
weight_delta_kg=Decimal("-50"),
outside_weight_after_kg=Decimal("70"),
source_doc_type="库存流水",
source_doc_id=102,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB20260614001",
created_at=self.now,
updated_at=self.now,
),
ProductionBatchLedgerTxn(
id=3,
production_ledger_id=1,
txn_type="生产余料入库",
qty_delta=Decimal("0"),
weight_delta_kg=Decimal("-18"),
outside_weight_after_kg=Decimal("52"),
source_doc_type="库存流水",
source_doc_id=103,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB20260614001",
created_at=self.now,
updated_at=self.now,
),
ProductionBatchLedgerTxn(
id=4,
production_ledger_id=1,
txn_type="生产废料入库",
qty_delta=Decimal("0"),
weight_delta_kg=Decimal("-6"),
outside_weight_after_kg=Decimal("46"),
source_doc_type="库存流水",
source_doc_id=104,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB20260614001",
created_at=self.now,
updated_at=self.now,
),
ProductionBatchLedgerTxn(
id=5,
production_ledger_id=1,
txn_type="结单核销",
qty_delta=Decimal("0"),
weight_delta_kg=Decimal("-4"),
outside_weight_after_kg=Decimal("42"),
source_doc_type="结单",
source_doc_id=105,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB20260614001",
created_at=self.now,
updated_at=self.now,
),
]
report = OperationReport(
id=1,
report_no="BG20260614001",
production_ledger_id=1,
employee_id=1,
report_source="MINIAPP",
start_time=self.now,
end_time=self.now,
report_qty=Decimal("100"),
good_qty=Decimal("96"),
scrap_qty=Decimal("4"),
rework_qty=Decimal("0"),
downtime_minutes=Decimal("0"),
extra_cost_amount=Decimal("0"),
is_abnormal=0,
created_at=self.now,
updated_at=self.now,
)
self.db.add_all([raw_item, product, warehouse, lot, ledger, dept, employee, *txns, report])
self.db.commit()
def _lot_read(self) -> DashboardMaterialLotRead:
return DashboardMaterialLotRead(
lot_id=1,
lot_no="YL0001",
material_item_id=1,
material_code="RM-001",
material_name="冷轧钢板",
warehouse_name="原材料库",
inbound_weight_kg=500,
remaining_weight_kg=300,
unit_cost=4,
inbound_amount=2000,
issued_batch_count=0,
issued_weight_kg=0,
finished_batch_count=0,
finished_qty=0,
delivery_count=0,
status="AVAILABLE",
quality_status="PASS",
created_at=self.now,
)
def test_production_ledger_graph_replaces_work_order_main_chain(self) -> None:
nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read())
node_types = {node.type for node in nodes}
self.assertIn("production_ledger", node_types)
self.assertIn("production_issue_txn", node_types)
self.assertIn("ledger_outside_balance", node_types)
self.assertIn("production_ledger_settlement", node_types)
self.assertIn("finished_inbound", node_types)
self.assertIn("surplus_inbound", node_types)
self.assertIn("scrap_inbound", node_types)
self.assertNotIn("work_order", node_types)
def test_production_ledger_title_stage_and_outside_balance_metric(self) -> None:
nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read())
ledger_node = next(node for node in nodes if node.type == "production_ledger")
metrics = {item.label: item.value for item in ledger_node.metrics}
self.assertEqual(ledger_node.title, "YL0001 · 钢制碗")
self.assertEqual(ledger_node.stage, "生产台账")
self.assertEqual(metrics["库外余额"], "42 kg")
def test_material_lot_dashboard_list_excludes_scrap_warehouse_lots(self) -> None:
scrap_warehouse = Warehouse(
id=3,
warehouse_code="WH-SCRAP",
warehouse_name="废料库",
warehouse_type="SCRAP",
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
scrap_lot = StockLot(
id=30,
lot_no="SCRAPI-YL0001-001",
lot_role="PRODUCTION_SCRAP",
item_id=1,
warehouse_id=3,
source_doc_type="PRODUCTION_SCRAP_IN",
source_doc_id=1,
inbound_qty=Decimal("0"),
inbound_weight_kg=Decimal("5"),
remaining_qty=Decimal("0"),
remaining_weight_kg=Decimal("5"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("4"),
quality_status="SCRAP",
status="AVAILABLE",
created_at=self.now,
updated_at=self.now,
)
self.db.add_all([scrap_warehouse, scrap_lot])
self.db.commit()
rows = list_dashboard_material_lots(limit=100, db=self.db)
lot_nos = {row.lot_no for row in rows}
self.assertIn("YL0001", lot_nos)
self.assertNotIn("SCRAPI-YL0001-001", lot_nos)
self.assertTrue(all(row.warehouse_name == "原材料库" for row in rows))
def test_operation_report_respects_smart_operation_report_switch(self) -> None:
nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read())
self.assertIn("operation_report", {node.type for node in nodes})
self.db.add(
SystemConfig(
config_code="SMART_OPERATION_REPORT_ENABLED",
config_name="对接智能报工小程序",
config_value="关闭",
status="ACTIVE",
created_at=self.now,
updated_at=self.now,
)
)
self.db.commit()
nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read())
self.assertNotIn("operation_report", {node.type for node in nodes})
def test_route_returns_new_graph_when_production_ledger_exists(self) -> None:
response = get_material_lot_lifecycle(1, self.db)
node_types = {node.type for node in response.nodes}
self.assertIn("production_ledger", node_types)
self.assertNotIn("work_order", node_types)
def test_route_legacy_fallback_does_not_crash_without_production_ledger(self) -> None:
legacy_lot = StockLot(
id=2,
lot_no="YL9999",
lot_role="RAW_MATERIAL",
item_id=1,
warehouse_id=1,
source_doc_type="期初入库",
source_doc_id=2,
inbound_qty=Decimal("0"),
inbound_weight_kg=Decimal("80"),
remaining_qty=Decimal("0"),
remaining_weight_kg=Decimal("80"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("4"),
quality_status="PASS",
status="AVAILABLE",
created_at=self.now,
updated_at=self.now,
)
self.db.add(legacy_lot)
self.db.commit()
response = get_material_lot_lifecycle(2, self.db)
self.assertEqual(response.lot.lot_no, "YL9999")
node_types = {node.type for node in response.nodes}
self.assertIn("raw_material_lot", node_types)
self.assertNotIn("production_ledger", node_types)
def test_delivery_edges_target_finished_inbound_for_exact_finished_lot(self) -> None:
finished_lot_a = StockLot(
id=10,
lot_no="FGL-A",
lot_role="INVENTORY",
item_id=2,
warehouse_id=1,
source_doc_type="PRODUCTION_LEDGER_IN",
source_doc_id=1,
inbound_qty=Decimal("40"),
inbound_weight_kg=Decimal("20"),
remaining_qty=Decimal("40"),
remaining_weight_kg=Decimal("20"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("8"),
quality_status="PASS",
status="AVAILABLE",
created_at=self.now,
updated_at=self.now,
)
finished_lot_b = StockLot(
id=11,
lot_no="FGL-B",
lot_role="INVENTORY",
item_id=2,
warehouse_id=1,
source_doc_type="PRODUCTION_LEDGER_IN",
source_doc_id=1,
inbound_qty=Decimal("60"),
inbound_weight_kg=Decimal("30"),
remaining_qty=Decimal("0"),
remaining_weight_kg=Decimal("0"),
locked_qty=Decimal("0"),
locked_weight_kg=Decimal("0"),
unit_cost=Decimal("8"),
quality_status="PASS",
status="AVAILABLE",
created_at=self.now,
updated_at=self.now,
)
finished_txn_a = ProductionBatchLedgerTxn(
id=20,
production_ledger_id=1,
txn_type="成品入库",
qty_delta=Decimal("40"),
weight_delta_kg=Decimal("-20"),
outside_weight_after_kg=Decimal("70"),
source_doc_type="库存流水",
source_doc_id=201,
source_line_id=10,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB-FGL-A",
created_at=self.now,
updated_at=self.now,
)
finished_txn_b = ProductionBatchLedgerTxn(
id=21,
production_ledger_id=1,
txn_type="成品入库",
qty_delta=Decimal("60"),
weight_delta_kg=Decimal("-30"),
outside_weight_after_kg=Decimal("42"),
source_doc_type="库存流水",
source_doc_id=202,
source_line_id=11,
biz_time=self.now,
operator_user_id=1,
remark="单据批次号：PB-FGL-B",
created_at=self.now,
updated_at=self.now,
)
delivery = Delivery(
id=1,
delivery_no="FH20260614001",
warehouse_id=1,
delivery_date=self.now,
status="POSTED",
created_at=self.now,
updated_at=self.now,
)
delivery_item = DeliveryItem(
id=1,
delivery_id=1,
line_no=1,
product_item_id=2,
lot_id=11,
delivery_qty=Decimal("10"),
delivery_weight_kg=Decimal("5"),
unit_price=Decimal("12"),
line_amount=Decimal("120"),
status="POSTED",
created_at=self.now,
updated_at=self.now,
)
self.db.add_all([finished_lot_a, finished_lot_b, finished_txn_a, finished_txn_b, delivery, delivery_item])
self.db.commit()
nodes, edges = build_material_lot_lifecycle_graph(self.db, self._lot_read())
delivery_node = next(node for node in nodes if node.type == "delivery")
finished_a = next(node for node in nodes if node.type == "finished_inbound" and node.subtitle == "PB-FGL-A")
finished_b = next(node for node in nodes if node.type == "finished_inbound" and node.subtitle == "PB-FGL-B")
delivery_source_ids = {edge.source for edge in edges if edge.target == delivery_node.id}
self.assertIn(finished_b.id, delivery_source_ids)
self.assertNotIn(finished_a.id, delivery_source_ids)
self.assertNotIn("ledger-1", delivery_source_ids)
def test_reopen_txn_uses_reopen_node_type_not_writeoff(self) -> None:
self.db.add(
ProductionBatchLedgerTxn(
id=30,
production_ledger_id=1,
txn_type="重新开工",
qty_delta=Decimal("0"),
weight_delta_kg=Decimal("0"),
outside_weight_after_kg=Decimal("42"),
source_doc_type="生产台账",
source_doc_id=1,
biz_time=self.now,
operator_user_id=1,
remark="重新开工",
created_at=self.now,
updated_at=self.now,
)
)
self.db.commit()
nodes, _ = build_material_lot_lifecycle_graph(self.db, self._lot_read())
reopen_node = next(node for node in nodes if node.title == "重新开工")
self.assertEqual(reopen_node.type, "ledger_reopen")
self.assertEqual(reopen_node.stage, "重新开工")
if __name__ == "__main__":
unittest.main()