311 lines
11 KiB
Python
311 lines
11 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 InventoryTxn, StockLot, WarehouseLocation # noqa: E402
|
|
|
|
|
|
class OpeningInventoryImportTest(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 = Warehouse(
|
|
id=1,
|
|
warehouse_code="WH-RAW",
|
|
warehouse_name="原材料仓",
|
|
warehouse_type="RAW",
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.finished_warehouse = Warehouse(
|
|
id=2,
|
|
warehouse_code="WH-FG",
|
|
warehouse_name="成品仓",
|
|
warehouse_type="FINISHED",
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.raw_location = WarehouseLocation(
|
|
id=1,
|
|
warehouse_id=1,
|
|
location_code="RAW-A",
|
|
location_name="原料主库位",
|
|
is_default=1,
|
|
is_locked=0,
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.finished_location = WarehouseLocation(
|
|
id=2,
|
|
warehouse_id=2,
|
|
location_code="FG-A",
|
|
location_name="成品主库位",
|
|
is_default=1,
|
|
is_locked=0,
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.raw_item = Item(
|
|
id=1,
|
|
item_code="原材料00001",
|
|
item_name="304不锈钢",
|
|
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.finished_item = Item(
|
|
id=2,
|
|
item_code="产品需规00001",
|
|
item_name="五金支架",
|
|
item_type="FINISHED_GOOD",
|
|
unit_weight_kg=Decimal("2.5"),
|
|
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_warehouse,
|
|
self.finished_warehouse,
|
|
self.raw_location,
|
|
self.finished_location,
|
|
self.raw_item,
|
|
self.finished_item,
|
|
])
|
|
self.db.commit()
|
|
|
|
def _workbook_bytes(self, rows: list[list[object]]) -> bytes:
|
|
from app.api.routes.inventory import OPENING_INVENTORY_HEADERS
|
|
|
|
workbook = Workbook()
|
|
worksheet = workbook.active
|
|
worksheet.title = "期初导入"
|
|
worksheet.append(OPENING_INVENTORY_HEADERS)
|
|
for row in rows:
|
|
worksheet.append(row)
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
return output.getvalue()
|
|
|
|
def test_opening_template_exports_expected_headers(self) -> None:
|
|
from app.api.routes.inventory import OPENING_INVENTORY_HEADERS, build_opening_inventory_template
|
|
|
|
workbook = build_opening_inventory_template(self.db, "FINISHED")
|
|
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 = list(rows[0])
|
|
|
|
self.assertEqual(headers, OPENING_INVENTORY_HEADERS)
|
|
self.assertEqual(len(rows), 3)
|
|
self.assertEqual(rows[1][headers.index("仓库类型")], "成品库")
|
|
self.assertEqual(rows[1][headers.index("批次号")], "示例-FG-0001")
|
|
self.assertEqual(rows[2][headers.index("批次号")], "示例-FG-0002")
|
|
|
|
def test_raw_opening_template_uses_weight_only_headers(self) -> None:
|
|
from app.api.routes.inventory import build_opening_inventory_template
|
|
|
|
workbook = build_opening_inventory_template(self.db, "RAW")
|
|
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 = list(rows[0])
|
|
|
|
self.assertNotIn("数量", headers)
|
|
self.assertIn("重量(kg)", headers)
|
|
self.assertEqual(rows[1][headers.index("仓库类型")], "原材料库")
|
|
self.assertEqual(rows[1][headers.index("重量(kg)")], 500)
|
|
|
|
def test_import_opening_template_ignores_sample_rows(self) -> None:
|
|
from app.api.routes.inventory import build_opening_inventory_template, import_opening_inventory_workbook
|
|
|
|
workbook = build_opening_inventory_template(self.db, "AUX")
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
|
|
with self.assertRaises(ValueError) as exc:
|
|
import_opening_inventory_workbook(self.db, "AUX", output.getvalue(), user_id=1)
|
|
|
|
self.assertIn("没有可导入的期初库存明细", str(exc.exception))
|
|
|
|
def test_import_opening_inventory_writes_finished_qty_weight_balance_and_txn(self) -> None:
|
|
from app.api.routes.inventory import import_opening_inventory_workbook
|
|
|
|
content = self._workbook_bytes(
|
|
[[
|
|
"成品库",
|
|
"成品仓",
|
|
"成品主库位",
|
|
"产品需规00001",
|
|
"五金支架",
|
|
"OPEN-FG-001",
|
|
12,
|
|
30,
|
|
8.5,
|
|
"初始化清点",
|
|
"期初成品",
|
|
]]
|
|
)
|
|
|
|
result = import_opening_inventory_workbook(self.db, "FINISHED", content, user_id=1)
|
|
|
|
lot = self.db.scalar(select(StockLot).where(StockLot.lot_no == "OPEN-FG-001"))
|
|
balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.finished_item.id))
|
|
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
|
|
|
self.assertEqual(result["imported"], 1)
|
|
self.assertEqual(float(lot.inbound_qty), 12)
|
|
self.assertEqual(float(lot.inbound_weight_kg), 30)
|
|
self.assertEqual(float(lot.remaining_qty), 12)
|
|
self.assertEqual(float(lot.remaining_weight_kg), 30)
|
|
self.assertEqual(float(balance.qty_on_hand), 12)
|
|
self.assertEqual(float(balance.weight_on_hand_kg), 30)
|
|
self.assertEqual(float(txn.qty_change), 12)
|
|
self.assertEqual(float(txn.weight_change_kg), 30)
|
|
self.assertEqual(txn.txn_type, "OPENING_IN")
|
|
|
|
def test_import_opening_inventory_keeps_raw_material_quantity_zero(self) -> None:
|
|
from app.api.routes.inventory import import_opening_inventory_workbook
|
|
|
|
content = self._workbook_bytes(
|
|
[[
|
|
"原材料库",
|
|
"原材料仓",
|
|
"原料主库位",
|
|
"原材料00001",
|
|
"304不锈钢",
|
|
"OPEN-RAW-001",
|
|
99,
|
|
500,
|
|
4.2,
|
|
"初始化清点",
|
|
"期初原材料",
|
|
]]
|
|
)
|
|
|
|
import_opening_inventory_workbook(self.db, "RAW", content, user_id=1)
|
|
|
|
lot = self.db.scalar(select(StockLot).where(StockLot.lot_no == "YL0001"))
|
|
balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.raw_item.id))
|
|
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
|
|
|
self.assertIsNotNone(lot)
|
|
self.assertEqual(float(lot.inbound_qty), 0)
|
|
self.assertEqual(float(lot.remaining_qty), 0)
|
|
self.assertEqual(float(balance.qty_on_hand), 0)
|
|
self.assertEqual(float(txn.qty_change), 0)
|
|
self.assertEqual(float(lot.inbound_weight_kg), 500)
|
|
|
|
def test_import_opening_inventory_generates_global_raw_lot_numbers(self) -> None:
|
|
from app.api.routes.inventory import import_opening_inventory_workbook
|
|
|
|
content = self._workbook_bytes(
|
|
[
|
|
[
|
|
"原材料库",
|
|
"原材料仓",
|
|
"原料主库位",
|
|
"原材料00001",
|
|
"304不锈钢",
|
|
"用户手填RAW-001",
|
|
0,
|
|
500,
|
|
4.2,
|
|
"初始化清点",
|
|
"期初原材料1",
|
|
],
|
|
[
|
|
"原材料库",
|
|
"原材料仓",
|
|
"原料主库位",
|
|
"原材料00001",
|
|
"304不锈钢",
|
|
"用户手填RAW-002",
|
|
0,
|
|
600,
|
|
4.2,
|
|
"初始化清点",
|
|
"期初原材料2",
|
|
],
|
|
]
|
|
)
|
|
|
|
import_opening_inventory_workbook(self.db, "RAW", content, user_id=1)
|
|
|
|
lot_nos = self.db.scalars(
|
|
select(StockLot.lot_no)
|
|
.where(
|
|
StockLot.item_id == self.raw_item.id,
|
|
StockLot.source_doc_type == "OPENING_STOCK",
|
|
)
|
|
.order_by(StockLot.id)
|
|
).all()
|
|
|
|
self.assertEqual(lot_nos, ["YL0001", "YL0002"])
|
|
|
|
def test_import_opening_inventory_accepts_raw_weight_only_template_without_quantity(self) -> None:
|
|
from app.api.routes.inventory import import_opening_inventory_workbook
|
|
|
|
workbook = Workbook()
|
|
worksheet = workbook.active
|
|
worksheet.title = "原材料库期初导入"
|
|
worksheet.append(["仓库类型", "仓库名称", "库位", "物料编码", "物料名称", "批次号", "重量(kg)", "单价", "来源说明", "备注"])
|
|
worksheet.append(["原材料库", "原材料仓", "原料主库位", "原材料00001", "304不锈钢", "OPEN-RAW-WEIGHT-ONLY", 600, 4.2, "初始化清点", "无数量列"])
|
|
output = BytesIO()
|
|
workbook.save(output)
|
|
|
|
import_opening_inventory_workbook(self.db, "RAW", output.getvalue(), user_id=1)
|
|
|
|
lot = self.db.scalar(select(StockLot).where(StockLot.lot_no == "YL0001"))
|
|
balance = self.db.scalar(select(StockBalance).where(StockBalance.item_id == self.raw_item.id))
|
|
|
|
self.assertIsNotNone(lot)
|
|
self.assertEqual(float(lot.inbound_qty), 0)
|
|
self.assertEqual(float(lot.remaining_qty), 0)
|
|
self.assertEqual(float(lot.inbound_weight_kg), 600)
|
|
self.assertEqual(float(balance.qty_on_hand), 0)
|
|
self.assertEqual(float(balance.weight_on_hand_kg), 600)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|