from __future__ import annotations import unittest from datetime import UTC, datetime from decimal import Decimal from io import BytesIO from openpyxl import Workbook, load_workbook 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.operations # noqa: E402,F401 from app.models.base import Base # noqa: E402 from app.models.master_data import Item, StockBalance, Warehouse # noqa: E402 from app.models.operations import StockLot, StocktakeLine, StocktakeWarehouse # noqa: E402 from app.services.stocktake import build_stocktake_workbook, start_stocktake # noqa: E402 class StocktakeExcelDiffConfirmTest(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._seed() def tearDown(self) -> None: self.db.close() def _seed(self) -> None: now = datetime.now(UTC) self.raw = Warehouse( id=1, warehouse_code="WH-RAW-01", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=now, updated_at=now, ) self.item = Item( id=1, item_code="原材料00001", item_name="冷轧钢板", item_type="RAW_MATERIAL", unit_weight_kg=Decimal("1"), safety_stock_weight_kg=Decimal("0"), scrap_sale_price=Decimal("0"), status="ACTIVE", created_at=now, updated_at=now, ) self.db.add_all([self.raw, self.item]) self.db.flush() self.lot = StockLot( id=1, lot_no="LOT-RAW-001", lot_role="INVENTORY", item_id=self.item.id, warehouse_id=self.raw.id, source_doc_type="OPENING", source_doc_id=1, inbound_qty=Decimal("0"), inbound_weight_kg=Decimal("100"), remaining_qty=Decimal("0"), remaining_weight_kg=Decimal("100"), locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=Decimal("2.5"), quality_status="PASS", status="AVAILABLE", created_at=now, updated_at=now, ) self.balance = StockBalance( item_id=self.item.id, warehouse_id=self.raw.id, qty_on_hand=Decimal("0"), weight_on_hand_kg=Decimal("100"), qty_available=Decimal("0"), weight_available_kg=Decimal("100"), qty_allocated=Decimal("0"), weight_allocated_kg=Decimal("0"), avg_unit_cost=Decimal("2.5"), created_at=now, updated_at=now, ) self.db.add_all([self.lot, self.balance]) self.db.commit() def test_empty_warehouse_stocktake_import_can_be_confirmed_and_unlocked(self) -> None: from app.services.stocktake import confirm_stocktake, import_stocktake_workbook now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() stocktake = start_stocktake(self.db, warehouse_ids=[aux.id], user_id=1, remark="空仓盘库") workbook = build_stocktake_workbook(self.db, stocktake.id) output = BytesIO() workbook.save(output) preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(preview["summary"]["match_count"], 0) self.assertEqual(preview["lines"], []) imported = self.db.get(type(stocktake), stocktake.id) self.assertEqual(imported.status, "IMPORTED") confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="确认盘库", remark="空仓确认") stocktake_warehouse = self.db.scalar( select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake.id) ) self.assertEqual(confirmed.status, "CONFIRMED") self.assertEqual(stocktake_warehouse.status, "UNLOCKED") def test_empty_warehouse_export_includes_stocktake_marker_row(self) -> None: now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() stocktake = start_stocktake(self.db, warehouse_ids=[aux.id], user_id=1, remark="空仓盘库") workbook = build_stocktake_workbook(self.db, stocktake.id) output = BytesIO() workbook.save(output) loaded = load_workbook(BytesIO(output.getvalue()), read_only=True, data_only=True) rows = list(loaded["辅料库"].iter_rows(values_only=True)) headers = rows[0] self.assertEqual(rows[1][headers.index("盘库单号")], stocktake.stocktake_no) self.assertEqual(rows[1][headers.index("仓库")], "辅料库") self.assertIn("空库", rows[1][headers.index("盘点备注")]) def test_empty_warehouse_marker_row_imports_as_clean_empty_preview(self) -> None: from openpyxl import Workbook from app.services.stocktake import import_stocktake_workbook now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() stocktake = start_stocktake(self.db, warehouse_ids=[aux.id], user_id=1, remark="空仓盘库") workbook = Workbook() worksheet = workbook.active worksheet.title = "辅料库" worksheet.append([ "盘库单号", "盘库校验码", "仓库ID", "快照行ID", "仓库", "库位", "物料编码", "物料名称", "批次号", "来源库存批次号", "系统数量", "系统重量(kg)", "系统单价", "盘点数量", "盘点重量(kg)", "盘点备注", ]) worksheet.append([ stocktake.stocktake_no, f"{stocktake.stocktake_no}|{aux.id}|AUX", aux.id, "", "辅料库", "", "", "", "", "", 0, 0, 0, 0, 0, "空库盘点校验行", ]) output = BytesIO() workbook.save(output) preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(preview["lines"], []) self.assertEqual(preview["summary"]["match_count"], 0) self.assertEqual(self.db.get(type(stocktake), stocktake.id).status, "IMPORTED") def test_exported_empty_warehouse_stocktake_can_be_confirmed_without_import_rows(self) -> None: from app.services.stocktake import confirm_stocktake now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() stocktake = start_stocktake(self.db, warehouse_ids=[aux.id], user_id=1, remark="空仓盘库") build_stocktake_workbook(self.db, stocktake.id) confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="确认盘库", remark="空仓确认") stocktake_warehouse = self.db.scalar( select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake.id) ) self.assertEqual(confirmed.status, "CONFIRMED") self.assertIsNotNone(confirmed.imported_at) self.assertEqual(stocktake_warehouse.status, "UNLOCKED") def test_confirm_stocktake_does_not_require_manual_confirm_text(self) -> None: from app.services.stocktake import confirm_stocktake now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() stocktake = start_stocktake(self.db, warehouse_ids=[aux.id], user_id=1, remark="空仓盘库") build_stocktake_workbook(self.db, stocktake.id) confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="", remark="空仓确认") self.assertEqual(confirmed.status, "CONFIRMED") def test_start_stocktake_locks_warehouse_and_snapshots_lots(self) -> None: stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库") self.assertEqual(stocktake.status, "LOCKED") line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id)) self.assertIsNotNone(line) self.assertEqual(line.lot_no, "LOT-RAW-001") self.assertEqual(float(line.snapshot_weight_kg), 100) self.assertEqual(line.diff_type, "MATCH") def test_start_stocktake_rejects_multiple_warehouses_for_current_bar_flow(self) -> None: from fastapi import HTTPException now = datetime.now(UTC) aux = Warehouse( id=2, warehouse_code="WH-AUX-01", warehouse_name="辅料库", warehouse_type="AUX", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add(aux) self.db.commit() with self.assertRaises(HTTPException) as exc: start_stocktake(self.db, warehouse_ids=[self.raw.id, aux.id], user_id=1, remark="多库盘库") self.assertEqual(exc.exception.status_code, 400) self.assertIn("一次只能盘一个仓库", exc.exception.detail) def test_export_workbook_has_one_sheet_per_selected_warehouse(self) -> None: stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库") workbook = build_stocktake_workbook(self.db, stocktake.id) output = BytesIO() workbook.save(output) loaded = load_workbook(BytesIO(output.getvalue()), read_only=True, data_only=True) self.assertEqual(loaded.sheetnames, ["原材料库"]) rows = list(loaded["原材料库"].iter_rows(values_only=True)) self.assertIn("快照行ID", rows[0]) self.assertIn("盘点重量(kg)", rows[0]) self.assertNotIn("系统数量", rows[0]) self.assertNotIn("盘点数量", rows[0]) self.assertEqual(rows[1][rows[0].index("批次号")], "LOT-RAW-001") def test_single_warehouse_export_contains_validation_marker(self) -> None: stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="原材料盘库") workbook = build_stocktake_workbook(self.db, stocktake.id) output = BytesIO() workbook.save(output) loaded = load_workbook(BytesIO(output.getvalue()), read_only=True, data_only=True) self.assertEqual(loaded.sheetnames, ["原材料库"]) rows = list(loaded["原材料库"].iter_rows(values_only=True)) headers = rows[0] self.assertIn("盘库校验码", headers) self.assertIn("仓库ID", headers) self.assertEqual(rows[1][headers.index("盘库单号")], stocktake.stocktake_no) self.assertEqual(rows[1][headers.index("仓库ID")], self.raw.id) self.assertEqual( rows[1][headers.index("盘库校验码")], f"{stocktake.stocktake_no}|{self.raw.id}|RAW", ) def _exported_workbook_bytes(self, stocktake) -> bytes: workbook = build_stocktake_workbook(self.db, stocktake.id) output = BytesIO() workbook.save(output) return output.getvalue() def test_import_rejects_workbook_from_other_stocktake_number(self) -> None: from app.services.stocktake import import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="原材料盘库") workbook = load_workbook(BytesIO(self._exported_workbook_bytes(stocktake))) worksheet = workbook["原材料库"] headers = [cell.value for cell in worksheet[1]] worksheet.cell(row=2, column=headers.index("盘库单号") + 1).value = "PK-OTHER-001" worksheet.cell(row=2, column=headers.index("盘库校验码") + 1).value = "PK-OTHER-001|1|RAW" output = BytesIO() workbook.save(output) with self.assertRaises(ValueError) as exc: import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertIn("不属于当前盘库单", str(exc.exception)) def test_import_rejects_workbook_from_other_warehouse(self) -> None: from app.services.stocktake import import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="原材料盘库") workbook = load_workbook(BytesIO(self._exported_workbook_bytes(stocktake))) worksheet = workbook["原材料库"] headers = [cell.value for cell in worksheet[1]] worksheet.cell(row=2, column=headers.index("仓库") + 1).value = "成品库" worksheet.cell(row=2, column=headers.index("仓库ID") + 1).value = 999 worksheet.cell(row=2, column=headers.index("盘库校验码") + 1).value = f"{stocktake.stocktake_no}|999|FINISHED" output = BytesIO() workbook.save(output) with self.assertRaises(ValueError) as exc: import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertIn("不属于当前仓库", str(exc.exception)) def test_reimport_overwrites_previous_stocktake_preview(self) -> None: from app.services.stocktake import import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="原材料盘库") workbook = load_workbook(BytesIO(self._exported_workbook_bytes(stocktake))) worksheet = workbook["原材料库"] headers = [cell.value for cell in worksheet[1]] worksheet.cell(row=2, column=headers.index("盘点重量(kg)") + 1).value = 90 output = BytesIO() workbook.save(output) first_preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(first_preview["summary"]["loss_count"], 1) workbook = load_workbook(BytesIO(self._exported_workbook_bytes(stocktake))) worksheet = workbook["原材料库"] headers = [cell.value for cell in worksheet[1]] worksheet.cell(row=2, column=headers.index("盘点重量(kg)") + 1).value = 100 output = BytesIO() workbook.save(output) second_preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(second_preview["summary"]["loss_count"], 0) self.assertEqual(second_preview["summary"]["match_count"], 1) line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id)) self.assertEqual(line.diff_type, "MATCH") self.assertEqual(float(line.counted_weight_kg), 100) def test_import_counted_workbook_marks_loss_gain_and_missing_rows(self) -> None: from app.services.stocktake import import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库") line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id)) workbook = Workbook() worksheet = workbook.active worksheet.title = "原材料库" worksheet.append([ "盘库单号", "盘库校验码", "仓库ID", "快照行ID", "仓库", "库位", "物料编码", "物料名称", "批次号", "来源库存批次号", "系统数量", "系统重量(kg)", "系统单价", "盘点数量", "盘点重量(kg)", "盘点备注", ]) worksheet.append([ stocktake.stocktake_no, f"{stocktake.stocktake_no}|{self.raw.id}|RAW", self.raw.id, line.id, "原材料库", "", "原材料00001", "冷轧钢板", "LOT-RAW-001", "", 0, 100, 2.5, 0, 90, "实盘少10kg", ]) output = BytesIO() workbook.save(output) preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(preview["summary"]["loss_count"], 1) loss_line = self.db.get(StocktakeLine, line.id) self.assertEqual(loss_line.diff_type, "LOSS") self.assertEqual(float(loss_line.diff_weight_kg), -10) self.assertEqual(loss_line.row_status, "READY") def test_confirm_stocktake_adjusts_lot_balance_and_writes_txn(self) -> None: from openpyxl import Workbook from app.models.operations import InventoryTxn, StocktakeAdjustment from app.services.stocktake import confirm_stocktake, import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库") line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id)) workbook = Workbook() worksheet = workbook.active worksheet.title = "原材料库" worksheet.append([ "盘库单号", "盘库校验码", "仓库ID", "快照行ID", "仓库", "库位", "物料编码", "物料名称", "批次号", "来源库存批次号", "系统数量", "系统重量(kg)", "系统单价", "盘点数量", "盘点重量(kg)", "盘点备注", ]) worksheet.append([ stocktake.stocktake_no, f"{stocktake.stocktake_no}|{self.raw.id}|RAW", self.raw.id, line.id, "原材料库", "", "原材料00001", "冷轧钢板", "LOT-RAW-001", "", 0, 100, 2.5, 0, 90, "实盘少10kg", ]) output = BytesIO() workbook.save(output) import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="确认盘库", remark="确认调整") self.assertEqual(confirmed.status, "CONFIRMED") self.db.refresh(self.lot) self.db.refresh(self.balance) self.assertEqual(float(self.lot.remaining_weight_kg), 90) self.assertEqual(float(self.balance.weight_on_hand_kg), 90) txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "STOCKTAKE_LOSS")) self.assertIsNotNone(txn) adjustment = self.db.scalar(select(StocktakeAdjustment).where(StocktakeAdjustment.stocktake_id == stocktake.id)) self.assertIsNotNone(adjustment) self.assertEqual(float(adjustment.weight_change_kg), -10) def test_stocktake_status_sequence_supports_status_driven_ui(self) -> None: from app.services.stocktake import confirm_stocktake, import_stocktake_workbook stocktake = start_stocktake(self.db, warehouse_ids=[self.raw.id], user_id=1, remark="月底盘库") self.assertEqual(stocktake.status, "LOCKED") workbook = build_stocktake_workbook(self.db, stocktake.id) self.assertEqual(self.db.get(type(stocktake), stocktake.id).status, "EXPORTED") output = BytesIO() workbook.save(output) preview = import_stocktake_workbook(self.db, stocktake.id, output.getvalue(), user_id=1) self.assertEqual(preview["stocktake"].status, "IMPORTED") self.assertEqual(preview["summary"]["match_count"], 1) confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="", remark="确认盘库") self.assertEqual(confirmed.status, "CONFIRMED") stocktake_warehouse = self.db.scalar( select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake.id) ) self.assertEqual(stocktake_warehouse.status, "UNLOCKED") if __name__ == "__main__": unittest.main()