511 lines
18 KiB
Python
511 lines
18 KiB
Python
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()
|