306 lines
12 KiB
Python
306 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import unittest
|
||
from datetime import datetime
|
||
from decimal import Decimal
|
||
from tempfile import TemporaryDirectory
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
from sqlalchemy import BigInteger, create_engine, select
|
||
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.models.base import Base # noqa: E402
|
||
from app.models.document_archive import DocumentArchive # noqa: E402
|
||
from app.models.master_data import Item, StockBalance, Warehouse # noqa: E402
|
||
from app.models.operations import InventoryTxn, StockLot, WarehouseLocation # noqa: E402
|
||
from app.api.routes.inventory import create_warehouse_inbound, create_warehouse_outbound # noqa: E402
|
||
from app.schemas.operations import WarehouseInboundCreate, WarehouseOutboundCreate # noqa: E402
|
||
from app.services.document_archives import ( # noqa: E402
|
||
ARCHIVE_STATUS_READY,
|
||
DOCUMENT_TYPE_WAREHOUSE_OPERATION,
|
||
collect_archive_context,
|
||
collect_warehouse_operation_archive_context,
|
||
generate_document_archive,
|
||
)
|
||
|
||
|
||
class WarehouseOperationDocumentArchiveTest(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)
|
||
|
||
def tearDown(self) -> None:
|
||
self.db.close()
|
||
|
||
def _seed_customer_supplied_txn(self) -> InventoryTxn:
|
||
warehouse = Warehouse(
|
||
id=1,
|
||
warehouse_code="RAW",
|
||
warehouse_name="原材料库",
|
||
warehouse_type="RAW",
|
||
status="ACTIVE",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
location = WarehouseLocation(
|
||
id=2,
|
||
warehouse_id=warehouse.id,
|
||
location_code="RAW-A",
|
||
location_name="原材料暂存位",
|
||
status="ACTIVE",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
item = Item(
|
||
id=3,
|
||
item_code="RM00001",
|
||
item_name="冷轧钢板",
|
||
item_type="RAW_MATERIAL",
|
||
specification="1.2mm",
|
||
unit_weight_kg=Decimal("1"),
|
||
status="ACTIVE",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
source_lot = StockLot(
|
||
id=4,
|
||
lot_no="SRC-YL0001",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=item.id,
|
||
warehouse_id=warehouse.id,
|
||
source_doc_type="期初入库",
|
||
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("0"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
lot = StockLot(
|
||
id=5,
|
||
lot_no="YL0001",
|
||
lot_role="RAW_MATERIAL",
|
||
item_id=item.id,
|
||
warehouse_id=warehouse.id,
|
||
location_id=location.id,
|
||
source_doc_type="客料入库",
|
||
source_doc_id=1,
|
||
source_material_lot_id=source_lot.id,
|
||
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("0"),
|
||
quality_status="PASS",
|
||
status="AVAILABLE",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
txn = InventoryTxn(
|
||
id=6,
|
||
txn_no="TXN-CUSTOMER-001",
|
||
txn_type="CUSTOMER_SUPPLIED_IN",
|
||
item_id=item.id,
|
||
warehouse_id=warehouse.id,
|
||
location_id=location.id,
|
||
lot_id=lot.id,
|
||
qty_change=Decimal("0"),
|
||
weight_change_kg=Decimal("100"),
|
||
unit_cost=Decimal("0"),
|
||
amount=Decimal("0"),
|
||
source_doc_type="客料入库",
|
||
source_doc_id=1,
|
||
logistics_waybill_no="SF123",
|
||
logistics_freight_amount=Decimal("12.5"),
|
||
logistics_photo_url="/uploads/demo.jpg",
|
||
biz_time=self.now,
|
||
remark="客户来料测试",
|
||
created_at=self.now,
|
||
updated_at=self.now,
|
||
)
|
||
balance = StockBalance(
|
||
id=7,
|
||
item_id=item.id,
|
||
warehouse_id=warehouse.id,
|
||
location_id=location.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("0"),
|
||
updated_at=self.now,
|
||
)
|
||
self.db.add_all([warehouse, location, item, source_lot, lot, txn, balance])
|
||
self.db.commit()
|
||
return txn
|
||
|
||
def test_collect_warehouse_operation_archive_context_uses_inventory_txn(self) -> None:
|
||
txn = self._seed_customer_supplied_txn()
|
||
|
||
context = collect_warehouse_operation_archive_context(self.db, txn.id)
|
||
|
||
self.assertEqual(context.document_type, DOCUMENT_TYPE_WAREHOUSE_OPERATION)
|
||
self.assertEqual(context.business_id, txn.id)
|
||
self.assertEqual(context.title, "客料入库单归档")
|
||
self.assertEqual(context.document_no, "客料入库-TXN-CUSTOMER-001")
|
||
self.assertEqual(context.partner_label, "仓库")
|
||
self.assertEqual(context.partner_name, "原材料库")
|
||
self.assertEqual(context.status, "客料入库")
|
||
self.assertEqual(context.address, "原材料库 / 原材料暂存位")
|
||
self.assertEqual(context.total_amount, Decimal("0"))
|
||
self.assertEqual(len(context.lines), 1)
|
||
line = context.lines[0]
|
||
self.assertEqual(line.item_code, "RM00001")
|
||
self.assertEqual(line.item_name, "冷轧钢板")
|
||
self.assertEqual(line.quantity, Decimal("100.000000"))
|
||
self.assertEqual(line.delivered_or_received_quantity, Decimal("100.000000"))
|
||
self.assertIn("库存流水号：TXN-CUSTOMER-001", line.remark or "")
|
||
self.assertIn("库存批次号：YL0001", line.remark or "")
|
||
self.assertIn("来源库存批次号：SRC-YL0001", line.remark or "")
|
||
self.assertIn("库位：原材料暂存位", line.remark or "")
|
||
self.assertIn("运单号：SF123", line.remark or "")
|
||
self.assertIn("运费：12.5", line.remark or "")
|
||
self.assertIn("辅助照片：已上传", line.remark or "")
|
||
self.assertIn("客户来料测试", line.remark or "")
|
||
|
||
def test_collect_archive_context_dispatches_warehouse_operation(self) -> None:
|
||
txn = self._seed_customer_supplied_txn()
|
||
|
||
context = collect_archive_context(self.db, DOCUMENT_TYPE_WAREHOUSE_OPERATION, txn.id)
|
||
|
||
self.assertEqual(context.document_type, DOCUMENT_TYPE_WAREHOUSE_OPERATION)
|
||
self.assertEqual(context.business_id, txn.id)
|
||
|
||
def test_unknown_english_txn_type_is_not_exposed_in_context(self) -> None:
|
||
txn = self._seed_customer_supplied_txn()
|
||
txn.txn_type = "unknown_code"
|
||
self.db.commit()
|
||
|
||
context = collect_warehouse_operation_archive_context(self.db, txn.id)
|
||
|
||
self.assertEqual(context.status, "未识别仓库操作")
|
||
self.assertEqual(context.title, "未识别仓库操作单归档")
|
||
self.assertEqual(context.document_no, "未识别仓库操作-TXN-CUSTOMER-001")
|
||
self.assertNotIn("unknown_code", context.remark)
|
||
self.assertNotIn("unknown_code", context.lines[0].remark or "")
|
||
|
||
def test_generate_document_archive_creates_pdf_for_warehouse_operation(self) -> None:
|
||
txn = self._seed_customer_supplied_txn()
|
||
|
||
with TemporaryDirectory() as tmp_dir, patch("app.services.document_archives.default_archive_root", return_value=tmp_dir):
|
||
result = generate_document_archive(self.db, DOCUMENT_TYPE_WAREHOUSE_OPERATION, txn.id)
|
||
|
||
archive = self.db.scalar(
|
||
select(DocumentArchive).where(
|
||
DocumentArchive.document_type == DOCUMENT_TYPE_WAREHOUSE_OPERATION,
|
||
DocumentArchive.business_id == txn.id,
|
||
)
|
||
)
|
||
self.assertIsNotNone(archive)
|
||
self.assertEqual(result.archive_status, ARCHIVE_STATUS_READY)
|
||
self.assertEqual(archive.status, ARCHIVE_STATUS_READY)
|
||
self.assertTrue(archive.file_name.endswith(".pdf"))
|
||
with open(archive.file_path, "rb") as pdf_file:
|
||
self.assertEqual(pdf_file.read(4), b"%PDF")
|
||
|
||
def test_customer_supplied_inbound_save_generates_warehouse_archive(self) -> None:
|
||
self._seed_customer_supplied_txn()
|
||
payload = WarehouseInboundCreate(
|
||
biz_type="CUSTOMER_SUPPLIED",
|
||
item_id=3,
|
||
warehouse_id=1,
|
||
location_id=2,
|
||
inbound_weight_kg=25,
|
||
inbound_qty=0,
|
||
unit_cost=0,
|
||
provider_name="测试客户",
|
||
waybill_no="SF999",
|
||
freight_amount=Decimal("8.5"),
|
||
remark="保存时生成仓库出入库单",
|
||
)
|
||
context = SimpleNamespace(user=SimpleNamespace(id=99))
|
||
|
||
with TemporaryDirectory() as tmp_dir, patch("app.services.document_archives.default_archive_root", return_value=tmp_dir):
|
||
result = create_warehouse_inbound(payload=payload, context=context, db=self.db)
|
||
|
||
self.assertEqual(result.archive_status, ARCHIVE_STATUS_READY)
|
||
self.assertEqual(result.archive_document_type, DOCUMENT_TYPE_WAREHOUSE_OPERATION)
|
||
self.assertIsNotNone(result.archive_business_id)
|
||
archive = self.db.scalar(
|
||
select(DocumentArchive).where(
|
||
DocumentArchive.document_type == DOCUMENT_TYPE_WAREHOUSE_OPERATION,
|
||
DocumentArchive.business_id == result.archive_business_id,
|
||
)
|
||
)
|
||
self.assertIsNotNone(archive)
|
||
self.assertTrue(archive.file_name.endswith(".pdf"))
|
||
|
||
def test_outsourcing_outbound_save_generates_warehouse_archive(self) -> None:
|
||
self._seed_customer_supplied_txn()
|
||
payload = WarehouseOutboundCreate(
|
||
biz_type="OUTSOURCING_OUT",
|
||
item_id=3,
|
||
warehouse_id=1,
|
||
location_id=2,
|
||
outbound_weight_kg=10,
|
||
outsourcing_party_name="测试委外厂",
|
||
waybill_no="SF-WW-001",
|
||
freight_amount=Decimal("9.9"),
|
||
remark="委外出库保存归档",
|
||
)
|
||
context = SimpleNamespace(user=SimpleNamespace(id=99))
|
||
|
||
with TemporaryDirectory() as tmp_dir, patch("app.services.document_archives.default_archive_root", return_value=tmp_dir):
|
||
result = create_warehouse_outbound(payload=payload, context=context, db=self.db)
|
||
|
||
self.assertEqual(result.archive_status, ARCHIVE_STATUS_READY)
|
||
self.assertEqual(result.archive_document_type, DOCUMENT_TYPE_WAREHOUSE_OPERATION)
|
||
self.assertIsNotNone(result.archive_business_id)
|
||
archive = self.db.scalar(
|
||
select(DocumentArchive).where(
|
||
DocumentArchive.document_type == DOCUMENT_TYPE_WAREHOUSE_OPERATION,
|
||
DocumentArchive.business_id == result.archive_business_id,
|
||
)
|
||
)
|
||
self.assertIsNotNone(archive)
|
||
self.assertTrue(archive.file_name.endswith(".pdf"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|