562 lines
22 KiB
Python
562 lines
22 KiB
Python
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()
|