from __future__ import annotations import unittest from datetime import date, datetime from decimal import Decimal from io import BytesIO from openpyxl import load_workbook from sqlalchemy import BigInteger, create_engine from sqlalchemy.ext.compiler import compiles from sqlalchemy.orm import Session, sessionmaker from sqlalchemy.pool import StaticPool @compiles(BigInteger, "sqlite") def _compile_big_integer_for_sqlite(type_, compiler, **kw) -> str: _ = type_, compiler, kw return "INTEGER" import app.models.document_archive # noqa: E402,F401 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.inventory import export_inventory_transaction_ledger # noqa: E402 from app.models.base import Base # noqa: E402 from app.models.document_archive import DocumentArchive # noqa: E402 from app.models.master_data import Item, Warehouse # noqa: E402 from app.models.operations import InventoryTxn, ProductionBatchLedger, ProductionBatchLedgerTxn, StockLot, WarehouseLocation # noqa: E402 from app.api.routes.inventory import list_inventory_transaction_ledger # noqa: E402 from app.services.operations import get_inventory_txn_ledger_query, get_inventory_txn_query # noqa: E402 class InventoryLedgerExportTest(unittest.TestCase): def setUp(self) -> None: engine = create_engine( "sqlite+pysqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool, 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, 11, 9, 0, 0) self._next_id = 100 self._seed_master_data() def tearDown(self) -> None: self.db.close() def _id(self) -> int: self._next_id += 1 return self._next_id def _seed_master_data(self) -> None: self.warehouse = Warehouse( id=1, warehouse_code="RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=self.now, updated_at=self.now, ) self.location = WarehouseLocation( id=2, warehouse_id=self.warehouse.id, location_code="RAW-A", location_name="原料主库位", is_default=1, status="ACTIVE", created_at=self.now, updated_at=self.now, ) self.item = Item( id=3, item_code="RM-LEDGER-001", item_name="测试冷轧钢", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), status="ACTIVE", created_at=self.now, updated_at=self.now, ) self.lot = StockLot( id=4, lot_no="YL0001", lot_role="RAW_MATERIAL", item_id=self.item.id, warehouse_id=self.warehouse.id, location_id=self.location.id, source_doc_type="期初入库", source_doc_id=1, inbound_qty=Decimal("0"), inbound_weight_kg=Decimal("500"), remaining_qty=Decimal("0"), remaining_weight_kg=Decimal("500"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("5.5"), quality_status="PASS", status="AVAILABLE", created_at=self.now, updated_at=self.now, ) self.db.add_all([self.warehouse, self.location, self.item, self.lot]) self.db.commit() def _seed_raw_inventory_txn( self, *, txn_type: str = "CUSTOMER_SUPPLIED_IN", source_doc_type: str = "客料入库", source_doc_id: int = 1, txn_no: str | None = None, biz_time: datetime | None = None, ) -> InventoryTxn: txn = InventoryTxn( id=self._id(), txn_no=txn_no or f"TXN-{self._next_id}", txn_type=txn_type, item_id=self.item.id, warehouse_id=self.warehouse.id, location_id=self.location.id, lot_id=self.lot.id, qty_change=Decimal("0"), weight_change_kg=Decimal("100"), unit_cost=Decimal("5.5"), amount=Decimal("550"), source_doc_type=source_doc_type, source_doc_id=source_doc_id, logistics_waybill_no="SF123", logistics_freight_amount=Decimal("12.5"), logistics_photo_url="/uploads/ledger.jpg", biz_time=biz_time or self.now, remark="流水导出测试备注", created_at=self.now, updated_at=self.now, ) self.db.add(txn) self.db.commit() return txn def test_inventory_ledger_returns_warehouse_archive_fields(self) -> None: txn = self._seed_raw_inventory_txn(txn_no="TXN-ARCHIVE-001") self.db.add( DocumentArchive( document_type="仓库出入库单", business_id=txn.id, document_no="客料入库-TXN-ARCHIVE-001", archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="客料入库.pdf", file_path="/tmp/客料入库.pdf", file_hash="abc", status="已归档", ) ) self.db.commit() row = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="RAW")).mappings().one() self.assertEqual(row["archive_status"], "已归档") self.assertEqual(row["archive_business_id"], txn.id) self.assertEqual(row["archive_document_type"], "仓库出入库单") self.assertIsNone(row["archive_error_message"]) def test_inventory_detail_transactions_return_archive_fields(self) -> None: txn = self._seed_raw_inventory_txn(txn_no="TXN-DETAIL-ARCHIVE") self.db.add( DocumentArchive( document_type="仓库出入库单", business_id=txn.id, document_no="客料入库-TXN-DETAIL-ARCHIVE", archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="客料入库.pdf", file_path="/tmp/客料入库.pdf", file_hash="detail", status="已归档", ) ) self.db.commit() row = self.db.execute(get_inventory_txn_query()).mappings().one() self.assertEqual(row["archive_status"], "已归档") self.assertEqual(row["archive_business_id"], txn.id) self.assertEqual(row["archive_document_type"], "仓库出入库单") self.assertIsNone(row["archive_error_message"]) def test_inventory_ledger_points_production_out_to_material_out_archive(self) -> None: self._seed_raw_inventory_txn( txn_type="PRODUCTION_OUT", source_doc_type="生产台账", source_doc_id=88, txn_no="TXN-PRODUCTION-OUT", ) row = self.db.execute(get_inventory_txn_ledger_query(warehouse_type="RAW")).mappings().one() self.assertEqual(row["archive_status"], "未生成") self.assertEqual(row["archive_document_type"], "生产领料出库单") self.assertEqual(row["archive_business_id"], 88) def test_transaction_ledger_export_uses_date_filter_and_chinese_headers(self) -> None: self._seed_raw_inventory_txn( txn_no="TXN-IN-RANGE", biz_time=datetime(2026, 6, 10, 8, 0, 0), ) self._seed_raw_inventory_txn( txn_no="TXN-OUT-RANGE", biz_time=datetime(2026, 5, 1, 8, 0, 0), ) response = export_inventory_transaction_ledger( warehouse_type="RAW", direction="IN", start_date=date(2026, 6, 1), end_date=date(2026, 6, 30), db=self.db, ) self.assertEqual(response.status_code, 200) self.assertIn( "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", response.media_type, ) workbook = load_workbook(BytesIO(response.body)) sheet = workbook.active headers = [cell.value for cell in sheet[1]] self.assertIn("业务时间", headers) self.assertIn("PDF归档状态", headers) rows = [[cell.value for cell in row] for row in sheet.iter_rows(min_row=2)] flattened = "\n".join(str(value) for row in rows for value in row) self.assertIn("TXN-IN-RANGE", flattened) self.assertNotIn("TXN-OUT-RANGE", flattened) def test_production_inbound_rows_share_settlement_archive_business_id(self) -> None: finished_txn_id = 301 surplus_txn = self._seed_raw_inventory_txn( txn_type="PRODUCTION_SURPLUS_IN", source_doc_type="生产台账", source_doc_id=66, txn_no="TXN-SURPLUS-SHARED", biz_time=datetime(2026, 6, 10, 10, 0, 0), ) ledger = ProductionBatchLedger( id=66, material_lot_id=self.lot.id, material_lot_no=self.lot.lot_no, material_item_id=self.item.id, product_item_id=self.item.id, status="在生产", total_issued_weight_kg=Decimal("100"), outside_weight_kg=Decimal("80"), created_at=self.now, updated_at=self.now, ) batch_no = "生产结算20260610100000000001" finished_ledger_txn = ProductionBatchLedgerTxn( id=401, production_ledger_id=ledger.id, txn_type="成品入库", qty_delta=Decimal("20"), weight_delta_kg=Decimal("-12"), outside_weight_after_kg=Decimal("88"), source_doc_type="库存流水", source_doc_id=finished_txn_id, biz_time=datetime(2026, 6, 10, 10, 0, 0), remark=f"成品入库联动生产台账；单据批次号：{batch_no}", created_at=self.now, updated_at=self.now, ) surplus_ledger_txn = ProductionBatchLedgerTxn( id=402, production_ledger_id=ledger.id, txn_type="生产余料入库", qty_delta=Decimal("0"), weight_delta_kg=Decimal("8"), outside_weight_after_kg=Decimal("80"), source_doc_type="库存流水", source_doc_id=surplus_txn.id, biz_time=datetime(2026, 6, 10, 10, 0, 0), remark=f"生产余料入库联动生产台账；单据批次号：{batch_no}", created_at=self.now, updated_at=self.now, ) self.db.add_all([ ledger, finished_ledger_txn, surplus_ledger_txn, DocumentArchive( document_type="生产入库结算单", business_id=finished_txn_id, document_no="生产入库结算-TXN-FINISHED-SHARED", archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="生产入库结算.pdf", file_path="/tmp/生产入库结算.pdf", file_hash="abc", status="已归档", ), ]) self.db.commit() result = list_inventory_transaction_ledger( warehouse_type="RAW", direction="IN", page=1, page_size=10, db=self.db, ) row = next(item for item in result.items if item.inventory_txn_id == surplus_txn.id) self.assertEqual(row.archive_document_type, "生产入库结算单") self.assertEqual(row.archive_business_id, finished_txn_id) self.assertEqual(row.archive_status, "已归档") if __name__ == "__main__": unittest.main()