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)