1167 lines
45 KiB
Python
1167 lines
45 KiB
Python
from __future__ import annotations
|
||
|
||
import unittest
|
||
from datetime import UTC, datetime
|
||
from decimal import Decimal
|
||
from io import BytesIO
|
||
|
||
from fastapi import HTTPException
|
||
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, StocktakeLine, WarehouseLocation # noqa: E402
|
||
from app.schemas.operations import WarehouseInboundCreate, WarehouseOutboundCreate # noqa: E402
|
||
|
||
|
||
class User:
|
||
id = 1
|
||
|
||
|
||
class Context:
|
||
user = User()
|
||
|
||
|
||
class ScrapWarehouseFlowTest(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.scrap_warehouse = Warehouse(
|
||
id=1,
|
||
warehouse_code="WH-SCRAP-01",
|
||
warehouse_name="废料库",
|
||
warehouse_type="SCRAP",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.raw_warehouse = Warehouse(
|
||
id=2,
|
||
warehouse_code="WH-RM-01",
|
||
warehouse_name="原材料库",
|
||
warehouse_type="RAW",
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.scrap_location = WarehouseLocation(
|
||
id=1,
|
||
warehouse_id=1,
|
||
location_code="SCR-E-01",
|
||
location_name="废料暂存位",
|
||
is_default=1,
|
||
is_locked=0,
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.raw_location = WarehouseLocation(
|
||
id=2,
|
||
warehouse_id=2,
|
||
location_code="RM-A-01",
|
||
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("1.6"),
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.product_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("1.2"),
|
||
status="ACTIVE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all(
|
||
[
|
||
self.scrap_warehouse,
|
||
self.raw_warehouse,
|
||
self.scrap_location,
|
||
self.raw_location,
|
||
self.raw_item,
|
||
self.product_item,
|
||
]
|
||
)
|
||
self.db.commit()
|
||
|
||
def _opening_workbook(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_scrap_opening_template_exports_sample_rows(self) -> None:
|
||
from app.api.routes.inventory import build_opening_inventory_template
|
||
|
||
workbook = build_opening_inventory_template(self.db, "SCRAP")
|
||
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(rows[1][headers.index("仓库类型")], "废料库")
|
||
self.assertEqual(rows[1][headers.index("仓库名称")], "废料库")
|
||
self.assertIn("示例-SCRAP-", rows[1][headers.index("批次号")])
|
||
self.assertGreater(rows[1][headers.index("重量(kg)")], 0)
|
||
|
||
def test_scrap_opening_import_is_weight_only(self) -> None:
|
||
from app.api.routes.inventory import import_opening_inventory_workbook
|
||
|
||
content = self._opening_workbook(
|
||
[[
|
||
"废料库",
|
||
"废料库",
|
||
"废料暂存位",
|
||
"原材料00001",
|
||
"304不锈钢边角料",
|
||
"OPEN-SCRAP-001",
|
||
99,
|
||
125,
|
||
1.6,
|
||
"初始化废料",
|
||
"期初废料",
|
||
]]
|
||
)
|
||
|
||
result = import_opening_inventory_workbook(self.db, "SCRAP", content, user_id=1)
|
||
|
||
lot = self.db.scalar(select(StockLot).where(StockLot.lot_no == "OPEN-SCRAP-001"))
|
||
balance = self.db.scalar(
|
||
select(StockBalance).where(
|
||
StockBalance.item_id == self.raw_item.id,
|
||
StockBalance.warehouse_id == self.scrap_warehouse.id,
|
||
)
|
||
)
|
||
|
||
self.assertIsNotNone(lot)
|
||
self.assertIsNotNone(balance)
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
self.assertIsNotNone(txn)
|
||
self.assertEqual(result["warehouse_type"], "SCRAP")
|
||
self.assertEqual(result["imported"], 1)
|
||
self.assertEqual(float(lot.inbound_qty), 0)
|
||
self.assertEqual(float(lot.remaining_qty), 0)
|
||
self.assertEqual(float(lot.inbound_weight_kg), 125)
|
||
self.assertEqual(float(balance.qty_on_hand), 0)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 125)
|
||
self.assertEqual(txn.txn_type, "OPENING_IN")
|
||
self.assertEqual(float(txn.qty_change), 0)
|
||
self.assertEqual(float(txn.weight_change_kg), 125)
|
||
|
||
def test_production_scrap_inbound_creates_scrap_lot_balance_and_txn(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound
|
||
|
||
with self.assertRaises(HTTPException) as cm:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="PRODUCTION_SCRAP_IN",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=36.5,
|
||
unit_cost=1.6,
|
||
source_material_sub_batch_no="JH_110501000185_0001",
|
||
source_material_summary="LAC-51012170/71左右后纵梁后段 1序报废边角料",
|
||
remark="生产废料入库",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(cm.exception.status_code, 400)
|
||
self.assertIn("生产废料入库必须选择生产工单", cm.exception.detail)
|
||
|
||
def test_production_surplus_returns_weight_to_selected_source_lot(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound
|
||
|
||
now = datetime.now(UTC)
|
||
source_lot = StockLot(
|
||
id=101,
|
||
lot_no="JH_5系钢板00002_0001",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30003,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("600"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("400"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("6"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("400"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("400"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("6"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([source_lot, balance])
|
||
self.db.commit()
|
||
|
||
with self.assertRaises(HTTPException) as cm:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="PRODUCTION_SURPLUS",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
inbound_weight_kg=80,
|
||
unit_cost=99,
|
||
source_lot_id=source_lot.id,
|
||
remark="生产余料归还",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.assertEqual(cm.exception.status_code, 400)
|
||
self.assertIn("生产余料入库必须选择生产工单", cm.exception.detail)
|
||
|
||
def test_generated_inventory_remarks_use_chinese_business_labels(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound
|
||
|
||
now = datetime.now(UTC)
|
||
source_lot = StockLot(
|
||
id=103,
|
||
lot_no="JH_5系钢板00002_0003",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30007,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("500"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("180"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("5"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("180"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("180"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("5"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([source_lot, balance])
|
||
self.db.commit()
|
||
|
||
with self.assertRaises(HTTPException) as surplus_exc:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="PRODUCTION_SURPLUS",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
inbound_weight_kg=20,
|
||
unit_cost=99,
|
||
source_lot_id=source_lot.id,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
outbound_weight_kg=15,
|
||
source_lots=[{"source_lot_id": source_lot.id, "outbound_weight_kg": 15}],
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
scrap_txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "SCRAP_OUT"))
|
||
|
||
self.assertEqual(surplus_exc.exception.status_code, 400)
|
||
self.assertIn("生产余料入库必须选择生产工单", surplus_exc.exception.detail)
|
||
self.assertIsNotNone(scrap_txn)
|
||
self.assertIn("报废出库", scrap_txn.remark)
|
||
self.assertNotIn("SCRAP_OUT", scrap_txn.remark)
|
||
|
||
def test_outsourcing_surplus_returns_weight_to_selected_source_lot(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound
|
||
|
||
now = datetime.now(UTC)
|
||
source_lot = StockLot(
|
||
id=102,
|
||
lot_no="JH_5系钢板00002_0002",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30006,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("500"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("120"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("5"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("120"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("120"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("5"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([source_lot, balance])
|
||
self.db.commit()
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_SURPLUS",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
inbound_weight_kg=30,
|
||
unit_cost=99,
|
||
source_lot_id=source_lot.id,
|
||
provider_name="委外厂A",
|
||
waybill_no="YD-WWSY-001",
|
||
freight_amount=12,
|
||
remark="委外余料归还",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.db.refresh(source_lot)
|
||
self.db.refresh(balance)
|
||
lots = self.db.scalars(select(StockLot).where(StockLot.item_id == self.raw_item.id)).all()
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.txn_type == "OUTSOURCING_SURPLUS_IN"))
|
||
|
||
self.assertEqual(row.lot_id, source_lot.id)
|
||
self.assertEqual(len(lots), 1)
|
||
self.assertEqual(float(source_lot.inbound_weight_kg), 500)
|
||
self.assertEqual(float(source_lot.remaining_weight_kg), 150)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 150)
|
||
self.assertEqual(float(balance.weight_available_kg), 150)
|
||
self.assertIsNotNone(txn)
|
||
self.assertEqual(txn.lot_id, source_lot.id)
|
||
self.assertEqual(float(txn.weight_change_kg), 30)
|
||
self.assertEqual(float(txn.unit_cost), 5)
|
||
self.assertEqual(float(txn.amount), 150)
|
||
self.assertIn("来源/提供方：委外厂A", txn.remark)
|
||
self.assertEqual(txn.logistics_waybill_no, "YD-WWSY-001")
|
||
self.assertEqual(float(txn.logistics_freight_amount), 12)
|
||
|
||
def test_outsourcing_outbound_deducts_selected_source_lots(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_outbound
|
||
|
||
now = datetime.now(UTC)
|
||
lot_a = StockLot(
|
||
id=201,
|
||
lot_no="JH_5系钢板00002_0001",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30001,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("100"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("50"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("5"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
lot_b = StockLot(
|
||
id=202,
|
||
lot_no="JH_5系钢板00002_0002",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30002,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("120"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("90"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("6"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
lot_c = StockLot(
|
||
id=203,
|
||
lot_no="JH_5系钢板00002_0003",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30003,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("80"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("60"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("7"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("200"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("200"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("6"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([lot_a, lot_b, lot_c, balance])
|
||
self.db.commit()
|
||
|
||
row = create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="OUTSOURCING_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
outbound_weight_kg=50,
|
||
source_lots=[
|
||
{"source_lot_id": lot_b.id, "outbound_weight_kg": 30},
|
||
{"source_lot_id": lot_c.id, "outbound_weight_kg": 20},
|
||
],
|
||
waybill_no="YD-WWCK-001",
|
||
freight_amount=18,
|
||
remark="委外出库指定批次",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.db.refresh(lot_a)
|
||
self.db.refresh(lot_b)
|
||
self.db.refresh(lot_c)
|
||
self.db.refresh(balance)
|
||
txns = self.db.scalars(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.txn_type == "OUTSOURCING_OUT")
|
||
.order_by(InventoryTxn.lot_id.asc())
|
||
).all()
|
||
|
||
self.assertEqual(row.lot_id, lot_b.id)
|
||
self.assertEqual(float(lot_a.remaining_weight_kg), 50)
|
||
self.assertEqual(float(lot_b.remaining_weight_kg), 60)
|
||
self.assertEqual(float(lot_c.remaining_weight_kg), 40)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 150)
|
||
self.assertEqual(float(balance.weight_available_kg), 150)
|
||
self.assertEqual([txn.lot_id for txn in txns], [lot_b.id, lot_c.id])
|
||
self.assertEqual([float(txn.weight_change_kg) for txn in txns], [-30, -20])
|
||
self.assertTrue(all(txn.logistics_waybill_no == "YD-WWCK-001" for txn in txns))
|
||
|
||
def test_scrap_outbound_deducts_selected_source_lots(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_outbound
|
||
|
||
now = datetime.now(UTC)
|
||
lot_a = StockLot(
|
||
id=211,
|
||
lot_no="JH_5系钢板00002_0001",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30011,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("100"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("40"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("5"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
lot_b = StockLot(
|
||
id=212,
|
||
lot_no="JH_5系钢板00002_0002",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=30012,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("130"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("75"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("6"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("115"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("115"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("5.5"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([lot_a, lot_b, balance])
|
||
self.db.commit()
|
||
|
||
row = create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
outbound_weight_kg=45,
|
||
source_lots=[
|
||
{"source_lot_id": lot_a.id, "outbound_weight_kg": 15},
|
||
{"source_lot_id": lot_b.id, "outbound_weight_kg": 30},
|
||
],
|
||
remark="报废出库指定批次",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.db.refresh(lot_a)
|
||
self.db.refresh(lot_b)
|
||
self.db.refresh(balance)
|
||
txns = self.db.scalars(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.txn_type == "SCRAP_OUT")
|
||
.order_by(InventoryTxn.lot_id.asc())
|
||
).all()
|
||
|
||
self.assertEqual(row.lot_id, lot_a.id)
|
||
self.assertEqual(float(lot_a.remaining_weight_kg), 25)
|
||
self.assertEqual(float(lot_b.remaining_weight_kg), 45)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 70)
|
||
self.assertEqual(float(balance.weight_available_kg), 70)
|
||
self.assertEqual([txn.lot_id for txn in txns], [lot_a.id, lot_b.id])
|
||
self.assertEqual([float(txn.weight_change_kg) for txn in txns], [-15, -30])
|
||
self.assertTrue(all("来源库存批次号" in (txn.remark or "") for txn in txns))
|
||
|
||
def test_outsourcing_scrap_inbound_accepts_product_item(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound
|
||
|
||
row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_SCRAP_IN",
|
||
item_id=self.product_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=12,
|
||
unit_cost=1.2,
|
||
provider_name="外协加工厂A",
|
||
waybill_no="YD-WWFL-001",
|
||
freight_amount=18.5,
|
||
remark="委外废料入库",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
lot = self.db.get(StockLot, row.lot_id)
|
||
self.assertIsNotNone(lot)
|
||
txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == lot.id))
|
||
|
||
self.assertIsNotNone(txn)
|
||
self.assertEqual(lot.lot_role, "OUTSOURCING_SCRAP")
|
||
self.assertEqual(lot.source_doc_type, "OUTSOURCING_SCRAP_IN")
|
||
self.assertEqual(float(lot.inbound_qty), 0)
|
||
self.assertEqual(lot.logistics_waybill_no, "YD-WWFL-001")
|
||
self.assertEqual(float(lot.logistics_freight_amount), 18.5)
|
||
self.assertEqual(txn.txn_type, "OUTSOURCING_SCRAP_IN")
|
||
self.assertEqual(txn.logistics_waybill_no, "YD-WWFL-001")
|
||
self.assertEqual(float(txn.logistics_freight_amount), 18.5)
|
||
|
||
def test_selected_generic_warehouse_operations_require_logistics_info(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound
|
||
|
||
with self.assertRaises(HTTPException) as inbound_exc:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_SCRAP_IN",
|
||
item_id=self.product_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=12,
|
||
unit_cost=1.2,
|
||
provider_name="外协加工厂A",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
self.assertEqual(inbound_exc.exception.status_code, 400)
|
||
self.assertIn("运单号", inbound_exc.exception.detail)
|
||
self.assertIn("运费", inbound_exc.exception.detail)
|
||
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OPENING",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=100,
|
||
unit_cost=1.6,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
with self.assertRaises(HTTPException) as outbound_exc:
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_SALE_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=40,
|
||
sale_unit_price=2.5,
|
||
buyer_name="废品回收商A",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
self.assertEqual(outbound_exc.exception.status_code, 400)
|
||
self.assertIn("运单号", outbound_exc.exception.detail)
|
||
self.assertIn("运费", outbound_exc.exception.detail)
|
||
|
||
def test_required_generic_logistics_allows_order_photo_instead_of_waybill(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound
|
||
|
||
inbound_row = create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OUTSOURCING_SCRAP_IN",
|
||
item_id=self.product_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=12,
|
||
unit_cost=1.2,
|
||
provider_name="外协加工厂A",
|
||
order_photo_url="/inventory/logistics-photos/test-inbound.jpg",
|
||
freight_amount=18.5,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
inbound_lot = self.db.get(StockLot, inbound_row.lot_id)
|
||
inbound_txn = self.db.scalar(select(InventoryTxn).where(InventoryTxn.lot_id == inbound_lot.id))
|
||
self.assertIsNone(inbound_lot.logistics_waybill_no)
|
||
self.assertEqual(inbound_lot.logistics_photo_url, "/inventory/logistics-photos/test-inbound.jpg")
|
||
self.assertEqual(inbound_txn.logistics_photo_url, "/inventory/logistics-photos/test-inbound.jpg")
|
||
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OPENING",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=100,
|
||
unit_cost=1.6,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_SALE_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=40,
|
||
sale_unit_price=2.5,
|
||
buyer_name="废品回收商A",
|
||
order_photo_url="/inventory/logistics-photos/test-outbound.jpg",
|
||
freight_amount=10,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
outbound_txn = self.db.scalar(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.txn_type == "SCRAP_SALE_OUT")
|
||
.order_by(InventoryTxn.id.desc())
|
||
.limit(1)
|
||
)
|
||
self.assertIsNone(outbound_txn.logistics_waybill_no)
|
||
self.assertEqual(outbound_txn.logistics_photo_url, "/inventory/logistics-photos/test-outbound.jpg")
|
||
|
||
def test_scrap_sale_outbound_schema_preserves_sale_fields(self) -> None:
|
||
payload = WarehouseOutboundCreate(
|
||
biz_type="SCRAP_SALE_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=40,
|
||
sale_unit_price=2.5,
|
||
buyer_name="废品回收商A",
|
||
sale_remark="按废不锈钢售卖",
|
||
)
|
||
|
||
self.assertEqual(float(payload.sale_unit_price), 2.5)
|
||
self.assertEqual(payload.buyer_name, "废品回收商A")
|
||
self.assertEqual(payload.sale_remark, "按废不锈钢售卖")
|
||
|
||
def test_scrap_sale_outbound_uses_sale_price_for_transaction_amount(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound
|
||
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="OPENING",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=100,
|
||
unit_cost=1.6,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
row = create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_SALE_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=40,
|
||
sale_unit_price=2.5,
|
||
buyer_name="废品回收商A",
|
||
waybill_no="YD-FLSM-001",
|
||
freight_amount=10,
|
||
sale_remark="按废不锈钢售卖",
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
txn = self.db.scalar(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.txn_type == "SCRAP_SALE_OUT")
|
||
.order_by(InventoryTxn.id.desc())
|
||
.limit(1)
|
||
)
|
||
balance = self.db.scalar(
|
||
select(StockBalance).where(
|
||
StockBalance.item_id == self.raw_item.id,
|
||
StockBalance.warehouse_id == self.scrap_warehouse.id,
|
||
)
|
||
)
|
||
|
||
self.assertIsNotNone(txn)
|
||
self.assertIsNotNone(balance)
|
||
self.assertEqual(row.warehouse_type, "SCRAP")
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 60)
|
||
self.assertEqual(float(txn.unit_cost), 2.5)
|
||
self.assertEqual(float(txn.amount), 100)
|
||
self.assertEqual(txn.logistics_waybill_no, "YD-FLSM-001")
|
||
self.assertEqual(float(txn.logistics_freight_amount), 10)
|
||
self.assertIn("废品回收商A", txn.remark)
|
||
|
||
def test_scrap_sale_outbound_deducts_selected_scrap_lots_and_records_raw_source_lots(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_outbound
|
||
|
||
now = datetime.now(UTC)
|
||
raw_source_lot = StockLot(
|
||
id=511,
|
||
lot_no="JH_304不锈钢00001_0001",
|
||
lot_role="INVENTORY",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.raw_warehouse.id,
|
||
location_id=self.raw_location.id,
|
||
source_doc_type="PURCHASE_RECEIPT",
|
||
source_doc_id=401,
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("100"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("20"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("6"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
scrap_lot_a = StockLot(
|
||
id=611,
|
||
lot_no="FL-20260602-001",
|
||
lot_role="SCRAP",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
source_doc_type="PRODUCTION_SCRAP_IN",
|
||
source_doc_id=501,
|
||
source_material_lot_id=raw_source_lot.id,
|
||
source_material_sub_batch_no="JH_旧文本来源_0009",
|
||
source_material_summary="来源库存批次号：JH_304不锈钢00001_0001",
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("40"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("25"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("1.2"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
scrap_lot_b = StockLot(
|
||
id=612,
|
||
lot_no="FL-20260602-002",
|
||
lot_role="SCRAP",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
source_doc_type="OUTSOURCING_SCRAP_IN",
|
||
source_doc_id=502,
|
||
source_material_sub_batch_no="JH_304不锈钢00001_0002",
|
||
source_material_summary="委外废料；来源库存批次号：JH_304不锈钢00001_0002",
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("60"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("45"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("1.3"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
balance = StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("70"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("70"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("1.25"),
|
||
updated_at=now,
|
||
)
|
||
self.db.add_all([raw_source_lot, scrap_lot_a, scrap_lot_b, balance])
|
||
self.db.commit()
|
||
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_SALE_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=45,
|
||
sale_unit_price=2.5,
|
||
buyer_name="废品回收商A",
|
||
waybill_no="YD-FLSM-002",
|
||
freight_amount=12,
|
||
source_lots=[
|
||
{"source_lot_id": scrap_lot_a.id, "outbound_weight_kg": 15},
|
||
{"source_lot_id": scrap_lot_b.id, "outbound_weight_kg": 30},
|
||
],
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
|
||
self.db.refresh(scrap_lot_a)
|
||
self.db.refresh(scrap_lot_b)
|
||
self.db.refresh(balance)
|
||
txns = self.db.scalars(
|
||
select(InventoryTxn)
|
||
.where(InventoryTxn.txn_type == "SCRAP_SALE_OUT")
|
||
.order_by(InventoryTxn.id.asc())
|
||
).all()
|
||
|
||
self.assertEqual(float(scrap_lot_a.remaining_weight_kg), 10)
|
||
self.assertEqual(float(scrap_lot_b.remaining_weight_kg), 15)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 25)
|
||
self.assertEqual(len(txns), 2)
|
||
self.assertIn("来源库存批次号 JH_304不锈钢00001_0001", txns[0].remark)
|
||
self.assertIn("来源库存批次号 JH_304不锈钢00001_0002", txns[1].remark)
|
||
|
||
def test_scrap_warehouse_rejects_non_scrap_operations(self) -> None:
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound
|
||
|
||
with self.assertRaises(HTTPException) as inbound_exc:
|
||
create_warehouse_inbound(
|
||
WarehouseInboundCreate(
|
||
biz_type="CUSTOMER_SUPPLIED",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
inbound_weight_kg=5,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
self.assertEqual(inbound_exc.exception.status_code, 400)
|
||
self.assertIn("废料库不允许执行该入库业务", inbound_exc.exception.detail)
|
||
|
||
with self.assertRaises(HTTPException) as outbound_exc:
|
||
create_warehouse_outbound(
|
||
WarehouseOutboundCreate(
|
||
biz_type="SCRAP_OUT",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
outbound_weight_kg=5,
|
||
),
|
||
context=Context(),
|
||
db=self.db,
|
||
)
|
||
self.assertEqual(outbound_exc.exception.status_code, 400)
|
||
self.assertIn("废料库不允许执行该出库业务", outbound_exc.exception.detail)
|
||
|
||
def test_stocktake_export_uses_scrap_sheet_name(self) -> None:
|
||
from app.services.stocktake import build_stocktake_workbook, start_stocktake
|
||
|
||
self.db.add(
|
||
StockLot(
|
||
id=10,
|
||
lot_no="SCRAP-STOCKTAKE-001",
|
||
lot_role="PRODUCTION_SCRAP",
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
source_doc_type="PRODUCTION_SCRAP_IN",
|
||
source_doc_id=10,
|
||
source_material_sub_batch_no="JH_110501000185_0001",
|
||
source_material_summary="盘库用废料",
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("66"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("66"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("1.6"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
)
|
||
)
|
||
self.db.add(
|
||
StockBalance(
|
||
item_id=self.raw_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("66"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("66"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("1.6"),
|
||
)
|
||
)
|
||
self.db.commit()
|
||
|
||
stocktake = start_stocktake(self.db, warehouse_ids=[self.scrap_warehouse.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.assertIn("废料库", loaded.sheetnames)
|
||
rows = list(loaded["废料库"].iter_rows(values_only=True))
|
||
headers = list(rows[0])
|
||
|
||
self.assertEqual(rows[1][headers.index("仓库")], "废料库")
|
||
self.assertEqual(rows[1][headers.index("库位")], "废料暂存位")
|
||
self.assertEqual(rows[1][headers.index("物料编码")], "原材料00001")
|
||
self.assertEqual(rows[1][headers.index("批次号")], "SCRAP-STOCKTAKE-001")
|
||
self.assertEqual(rows[1][headers.index("来源库存批次号")], "JH_110501000185_0001")
|
||
self.assertNotIn("系统数量", headers)
|
||
self.assertNotIn("盘点数量", headers)
|
||
self.assertEqual(rows[1][headers.index("系统重量(kg)")], 66)
|
||
self.assertEqual(rows[1][headers.index("系统单价")], 1.6)
|
||
|
||
def test_runtime_seed_normalizes_existing_scrap_warehouse_name(self) -> None:
|
||
from app.api.routes.master_data import _ensure_jiaheng_warehouses
|
||
|
||
self.scrap_warehouse.warehouse_name = "废料仓"
|
||
self.scrap_warehouse.remark = "旧废料仓说明"
|
||
self.db.add(self.scrap_warehouse)
|
||
self.db.commit()
|
||
|
||
_ensure_jiaheng_warehouses(self.db)
|
||
self.db.refresh(self.scrap_warehouse)
|
||
|
||
self.assertEqual(self.scrap_warehouse.warehouse_name, "废料库")
|
||
self.assertEqual(self.scrap_warehouse.remark, "生产废料、边角料、委外废料、报废品等废料库存")
|
||
|
||
|
||
def test_scrap_stocktake_keeps_product_scrap_weight_only_after_confirm(self) -> None:
|
||
from app.services.stocktake import build_stocktake_workbook, confirm_stocktake, import_stocktake_workbook, start_stocktake
|
||
|
||
self.db.add(
|
||
StockLot(
|
||
id=11,
|
||
lot_no="SCRAP-PRODUCT-001",
|
||
lot_role="OUTSOURCING_SCRAP",
|
||
item_id=self.product_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
source_doc_type="OUTSOURCING_SCRAP_IN",
|
||
source_doc_id=11,
|
||
source_material_summary="产品报废件盘库",
|
||
inbound_qty=Decimal("0"),
|
||
inbound_weight_kg=Decimal("30"),
|
||
remaining_qty=Decimal("0"),
|
||
remaining_weight_kg=Decimal("30"),
|
||
locked_qty=Decimal("0"),
|
||
locked_weight_kg=Decimal("0"),
|
||
unit_cost=Decimal("1.2"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
)
|
||
)
|
||
self.db.add(
|
||
StockBalance(
|
||
item_id=self.product_item.id,
|
||
warehouse_id=self.scrap_warehouse.id,
|
||
location_id=self.scrap_location.id,
|
||
qty_on_hand=Decimal("0"),
|
||
weight_on_hand_kg=Decimal("30"),
|
||
qty_available=Decimal("0"),
|
||
weight_available_kg=Decimal("30"),
|
||
qty_allocated=Decimal("0"),
|
||
weight_allocated_kg=Decimal("0"),
|
||
avg_unit_cost=Decimal("1.2"),
|
||
)
|
||
)
|
||
self.db.commit()
|
||
|
||
stocktake = start_stocktake(self.db, warehouse_ids=[self.scrap_warehouse.id], user_id=1, remark="产品废料盘库")
|
||
workbook = build_stocktake_workbook(self.db, stocktake.id)
|
||
output = BytesIO()
|
||
workbook.save(output)
|
||
loaded = load_workbook(BytesIO(output.getvalue()))
|
||
worksheet = loaded["废料库"]
|
||
headers = [cell.value for cell in worksheet[1]]
|
||
self.assertNotIn("盘点数量", headers)
|
||
worksheet.cell(row=2, column=headers.index("盘点重量(kg)") + 1, value=45)
|
||
imported_output = BytesIO()
|
||
loaded.save(imported_output)
|
||
|
||
import_stocktake_workbook(self.db, stocktake.id, imported_output.getvalue(), user_id=1)
|
||
confirmed = confirm_stocktake(self.db, stocktake.id, user_id=2, confirm_text="", remark="确认产品废料盘库")
|
||
|
||
self.assertEqual(confirmed.status, "CONFIRMED")
|
||
lot = self.db.scalar(select(StockLot).where(StockLot.lot_no == "SCRAP-PRODUCT-001"))
|
||
balance = self.db.scalar(
|
||
select(StockBalance).where(
|
||
StockBalance.item_id == self.product_item.id,
|
||
StockBalance.warehouse_id == self.scrap_warehouse.id,
|
||
)
|
||
)
|
||
line = self.db.scalar(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake.id))
|
||
|
||
self.assertIsNotNone(lot)
|
||
self.assertIsNotNone(balance)
|
||
self.assertIsNotNone(line)
|
||
self.assertEqual(float(line.counted_qty), 0)
|
||
self.assertEqual(float(line.diff_qty), 0)
|
||
self.assertEqual(float(lot.remaining_qty), 0)
|
||
self.assertEqual(float(lot.remaining_weight_kg), 45)
|
||
self.assertEqual(float(balance.qty_on_hand), 0)
|
||
self.assertEqual(float(balance.qty_available), 0)
|
||
self.assertEqual(float(balance.weight_on_hand_kg), 45)
|
||
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|