372 lines
14 KiB
Python
372 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import datetime
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import BigInteger, create_engine, select
|
|
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.models.base import Base # noqa: E402
|
|
from app.models.master_data import Bom, BomItem, Item, Warehouse # noqa: E402
|
|
from app.models.operations import ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot # noqa: E402
|
|
|
|
|
|
class ProductionBatchLedgerModelTest(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, 8, 8, 30)
|
|
|
|
def tearDown(self) -> None:
|
|
self.db.close()
|
|
|
|
def _seed_basic_items_and_lot(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,
|
|
)
|
|
self.db.add_all([raw_item, product, warehouse, lot])
|
|
self.db.flush()
|
|
|
|
def _seed_basic_items_bom_and_lot(self) -> None:
|
|
self._seed_basic_items_and_lot()
|
|
bom = Bom(
|
|
id=1,
|
|
bom_code="BOM-001",
|
|
product_item_id=2,
|
|
version_no="V1.0",
|
|
is_default=1,
|
|
status="ACTIVE",
|
|
created_at=self.now,
|
|
updated_at=self.now,
|
|
)
|
|
bom_item = BomItem(
|
|
id=1,
|
|
bom_id=1,
|
|
seq_no=1,
|
|
material_item_id=1,
|
|
usage_type="WEIGHT",
|
|
gross_weight_per_piece_kg=Decimal("0.6"),
|
|
net_weight_per_piece_kg=Decimal("0.5"),
|
|
loss_rate=Decimal("0"),
|
|
created_at=self.now,
|
|
updated_at=self.now,
|
|
)
|
|
self.db.add_all([bom, bom_item])
|
|
self.db.flush()
|
|
|
|
def _seed_ledger(
|
|
self,
|
|
*,
|
|
total_issued_weight_kg: Decimal,
|
|
outside_weight_kg: Decimal | None = None,
|
|
) -> ProductionBatchLedger:
|
|
ledger = ProductionBatchLedger(
|
|
material_lot_id=1,
|
|
material_lot_no="YL0001",
|
|
material_item_id=1,
|
|
product_item_id=2,
|
|
status="在生产",
|
|
miniapp_selectable=1,
|
|
total_issued_weight_kg=total_issued_weight_kg,
|
|
outside_weight_kg=outside_weight_kg if outside_weight_kg is not None else total_issued_weight_kg,
|
|
finished_inbound_qty=Decimal("0"),
|
|
surplus_inbound_weight_kg=Decimal("0"),
|
|
scrap_inbound_weight_kg=Decimal("0"),
|
|
writeoff_weight_kg=Decimal("0"),
|
|
first_issue_time=self.now,
|
|
last_issue_time=self.now,
|
|
created_at=self.now,
|
|
updated_at=self.now,
|
|
)
|
|
self.db.add(ledger)
|
|
self.db.flush()
|
|
return ledger
|
|
|
|
def test_models_persist_chinese_status_and_business_key(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,
|
|
)
|
|
self.db.add_all([raw_item, product, warehouse, lot])
|
|
self.db.flush()
|
|
|
|
ledger = ProductionBatchLedger(
|
|
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("200"),
|
|
outside_weight_kg=Decimal("200"),
|
|
finished_inbound_qty=Decimal("0"),
|
|
surplus_inbound_weight_kg=Decimal("0"),
|
|
scrap_inbound_weight_kg=Decimal("0"),
|
|
writeoff_weight_kg=Decimal("0"),
|
|
first_issue_time=self.now,
|
|
last_issue_time=self.now,
|
|
created_at=self.now,
|
|
updated_at=self.now,
|
|
)
|
|
self.db.add(ledger)
|
|
self.db.flush()
|
|
self.db.add(
|
|
ProductionBatchLedgerTxn(
|
|
production_ledger_id=ledger.id,
|
|
txn_type="生产出库",
|
|
qty_delta=Decimal("0"),
|
|
weight_delta_kg=Decimal("200"),
|
|
outside_weight_after_kg=Decimal("200"),
|
|
source_doc_type="库存流水",
|
|
source_doc_id=10,
|
|
biz_time=self.now,
|
|
operator_user_id=1,
|
|
remark="生产出库累计到生产台账",
|
|
created_at=self.now,
|
|
updated_at=self.now,
|
|
)
|
|
)
|
|
self.db.commit()
|
|
|
|
saved = self.db.scalar(select(ProductionBatchLedger).where(ProductionBatchLedger.material_lot_no == "YL0001"))
|
|
self.assertIsNotNone(saved)
|
|
self.assertEqual(saved.status, "在生产")
|
|
self.assertEqual(saved.miniapp_selectable, 1)
|
|
self.assertEqual(float(saved.outside_weight_kg), 200)
|
|
txns = self.db.scalars(select(ProductionBatchLedgerTxn)).all()
|
|
self.assertEqual(txns[0].txn_type, "生产出库")
|
|
|
|
def test_record_issue_accumulates_same_lot_and_product(self) -> None:
|
|
from app.services.production_batch_ledger import record_production_issue_to_ledger
|
|
|
|
self._seed_basic_items_and_lot()
|
|
first = record_production_issue_to_ledger(
|
|
self.db,
|
|
material_lot_id=1,
|
|
product_item_id=2,
|
|
issued_weight_kg=Decimal("120"),
|
|
operator_user_id=1,
|
|
source_doc_type="库存流水",
|
|
source_doc_id=100,
|
|
source_line_id=1001,
|
|
biz_time=datetime(2026, 6, 8, 9, 0),
|
|
remark="第一次出库",
|
|
)
|
|
second = record_production_issue_to_ledger(
|
|
self.db,
|
|
material_lot_id=1,
|
|
product_item_id=2,
|
|
issued_weight_kg=Decimal("80"),
|
|
operator_user_id=1,
|
|
source_doc_type="库存流水",
|
|
source_doc_id=101,
|
|
source_line_id=1002,
|
|
biz_time=datetime(2026, 6, 8, 14, 0),
|
|
remark="第二次出库",
|
|
)
|
|
self.db.commit()
|
|
|
|
self.assertEqual(first.id, second.id)
|
|
self.assertEqual(second.status, "在生产")
|
|
self.assertEqual(float(second.total_issued_weight_kg), 200)
|
|
self.assertEqual(float(second.outside_weight_kg), 200)
|
|
txns = self.db.scalars(select(ProductionBatchLedgerTxn).order_by(ProductionBatchLedgerTxn.id)).all()
|
|
self.assertEqual([row.txn_type for row in txns], ["生产出库", "生产出库"])
|
|
|
|
def test_lock_and_reopen_same_lot_product(self) -> None:
|
|
from app.services.production_batch_ledger import lock_production_batch_ledger, record_production_issue_to_ledger
|
|
|
|
self._seed_basic_items_and_lot()
|
|
ledger = record_production_issue_to_ledger(
|
|
self.db,
|
|
material_lot_id=1,
|
|
product_item_id=2,
|
|
issued_weight_kg=Decimal("200"),
|
|
operator_user_id=1,
|
|
source_doc_type="库存流水",
|
|
source_doc_id=100,
|
|
source_line_id=1001,
|
|
biz_time=datetime(2026, 6, 8, 9, 0),
|
|
remark="生产出库",
|
|
)
|
|
locked = lock_production_batch_ledger(
|
|
self.db,
|
|
production_ledger_id=ledger.id,
|
|
operator_user_id=1,
|
|
biz_time=datetime(2026, 6, 8, 18, 0),
|
|
remark="现场确认该批材料已无库外余量",
|
|
)
|
|
self.assertEqual(locked.status, "锁单")
|
|
self.assertEqual(locked.miniapp_selectable, 0)
|
|
self.assertEqual(float(locked.outside_weight_kg), 0)
|
|
self.assertEqual(float(locked.writeoff_weight_kg), 200)
|
|
|
|
reopened = record_production_issue_to_ledger(
|
|
self.db,
|
|
material_lot_id=1,
|
|
product_item_id=2,
|
|
issued_weight_kg=Decimal("50"),
|
|
operator_user_id=1,
|
|
source_doc_type="库存流水",
|
|
source_doc_id=102,
|
|
source_line_id=1003,
|
|
biz_time=datetime(2026, 6, 9, 9, 0),
|
|
remark="继续出库生产",
|
|
)
|
|
|
|
self.assertEqual(reopened.id, locked.id)
|
|
self.assertEqual(reopened.status, "在生产")
|
|
self.assertEqual(reopened.miniapp_selectable, 1)
|
|
self.assertEqual(float(reopened.total_issued_weight_kg), 250)
|
|
self.assertEqual(float(reopened.outside_weight_kg), 50)
|
|
txns = self.db.scalars(select(ProductionBatchLedgerTxn).order_by(ProductionBatchLedgerTxn.id)).all()
|
|
self.assertEqual([row.txn_type for row in txns], ["生产出库", "结单核销", "该批材料结单", "重新开工", "生产出库"])
|
|
|
|
def test_disabled_mode_calculates_from_finished_inbound_and_bom(self) -> None:
|
|
from app.services.production_batch_ledger import calculate_production_batch_ledger_snapshot
|
|
|
|
self._seed_basic_items_bom_and_lot()
|
|
ledger = self._seed_ledger(total_issued_weight_kg=Decimal("200"))
|
|
ledger.finished_inbound_qty = Decimal("100")
|
|
snapshot = calculate_production_batch_ledger_snapshot(self.db, ledger.id, smart_reporting_enabled=False)
|
|
|
|
self.assertEqual(float(snapshot.reference_finished_qty), 333)
|
|
self.assertEqual(float(snapshot.used_material_weight_kg), 60)
|
|
self.assertEqual(float(snapshot.theoretical_scrap_weight_kg), 10)
|
|
self.assertEqual(float(snapshot.theoretical_surplus_weight_kg), 140)
|
|
|
|
def test_lock_writes_off_remaining_outside_weight_after_inbounds(self) -> None:
|
|
from app.services.production_batch_ledger import lock_production_batch_ledger, post_finished_inbound_to_ledger
|
|
|
|
self._seed_basic_items_bom_and_lot()
|
|
ledger = self._seed_ledger(total_issued_weight_kg=Decimal("200"), outside_weight_kg=Decimal("200"))
|
|
post_finished_inbound_to_ledger(
|
|
self.db,
|
|
production_ledger_id=ledger.id,
|
|
finished_qty=Decimal("100"),
|
|
operator_user_id=1,
|
|
source_doc_type="成品入库",
|
|
source_doc_id=1,
|
|
source_line_id=1,
|
|
biz_time=datetime(2026, 6, 8, 12, 0),
|
|
remark="成品入库",
|
|
)
|
|
locked = lock_production_batch_ledger(
|
|
self.db,
|
|
production_ledger_id=ledger.id,
|
|
operator_user_id=1,
|
|
biz_time=datetime(2026, 6, 8, 18, 0),
|
|
remark="该批材料结单",
|
|
)
|
|
self.assertEqual(locked.status, "锁单")
|
|
self.assertEqual(float(locked.outside_weight_kg), 0)
|
|
self.assertEqual(float(locked.finished_inbound_qty), 100)
|