from __future__ import annotations import unittest from datetime import date, datetime from unittest.mock import patch from sqlalchemy import BigInteger, create_engine, select from sqlalchemy.ext.compiler import compiles from sqlalchemy.orm import Session, sessionmaker @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.api.routes.purchase import create_purchase_order, create_purchase_receipt, lock_purchase_order # noqa: E402 from app.models.base import Base # noqa: E402 from app.models.document_archive import DocumentArchive # noqa: E402 from app.models.master_data import Item, Material, Warehouse # noqa: E402 from app.models.operations import PurchaseOrderItem, Supplier # noqa: E402 from app.models.org import Department, Employee, User # noqa: E402 from app.schemas.document_archives import DocumentArchiveGenerateResult # noqa: E402 from app.schemas.operations import ( # noqa: E402 PurchaseOrderCreate, PurchaseOrderItemCreate, PurchaseReceiptCreate, PurchaseReceiptItemCreate, ) from app.services.auth import AuthContext # noqa: E402 class PurchaseReceiptDocumentArchiveTest(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_master_data() def tearDown(self) -> None: self.db.close() def _seed_master_data(self) -> None: now = datetime(2026, 6, 11, 9, 0, 0) self.department = Department( id=1, dept_code="D-WH-001", dept_name="仓储部", dept_type="WAREHOUSE", status="ACTIVE", created_at=now, updated_at=now, ) self.employee = Employee( id=1, employee_code="EMP-WH-001", employee_name="仓库接收员", dept_id=1, mobile="13800000000", status="ACTIVE", created_at=now, updated_at=now, ) self.user = User( id=1, username="receipt_tester", password_hash="x", employee_id=1, dept_id=1, nickname="到货测试员", is_super_admin=1, status="ACTIVE", created_at=now, updated_at=now, ) self.supplier = Supplier( id=1, supplier_code="SUP-RC-001", supplier_name="到货供应商", short_name="到货供应商", lead_time_days=0, default_tax_rate=0, status="ACTIVE", created_at=now, updated_at=now, ) self.material_item = Item( id=10, item_code="RM-RC-001", item_name="到货原材料", item_type="RAW_MATERIAL", specification="A-200", material_grade="45#", unit_weight_kg=1, status="ACTIVE", created_at=now, updated_at=now, ) self.material = Material( id=100, item_id=self.material_item.id, default_supplier_id=self.supplier.id, purchase_calc_mode="WEIGHT", min_purchase_qty=0, purchase_multiple_qty=0, purchase_multiple_weight_kg=0, lead_time_days=0, quality_rule=None, created_at=now, updated_at=now, ) self.warehouse = Warehouse( id=1, warehouse_code="WH-RAW", warehouse_name="原材料库", warehouse_type="RAW", status="ACTIVE", created_at=now, updated_at=now, ) self.db.add_all([ self.department, self.employee, self.user, self.supplier, self.material_item, self.material, self.warehouse, ]) self.db.commit() def _auth_context(self) -> AuthContext: return AuthContext( user=self.user, employee=self.employee, department=self.department, role_codes=["ADMIN"], role_names=["管理员"], permission_codes=[], ) def _create_locked_purchase_order(self) -> tuple[int, PurchaseOrderItem]: with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch( "app.api.routes.purchase.generate_document_archive", return_value=DocumentArchiveGenerateResult( business_id=1, document_type="采购订单", document_no="采购2026-00001", archive_status="已归档", archive_version=1, archive_error_message=None, ), ), ): order = create_purchase_order( PurchaseOrderCreate( supplier_id=self.supplier.id, order_date=date(2026, 6, 11), expected_date=date(2026, 6, 20), target_warehouse_type="RAW", items=[ PurchaseOrderItemCreate( material_item_id=self.material_item.id, order_weight_kg=120, unit_price=6.8, ) ], ), db=self.db, ) lock_purchase_order(order.purchase_order_id, db=self.db) po_item = self.db.scalar(select(PurchaseOrderItem).where(PurchaseOrderItem.purchase_order_id == order.purchase_order_id)) self.assertIsNotNone(po_item) return int(order.purchase_order_id), po_item def _receipt_payload(self, purchase_order_id: int, po_item: PurchaseOrderItem) -> PurchaseReceiptCreate: return PurchaseReceiptCreate( purchase_order_id=purchase_order_id, warehouse_id=self.warehouse.id, receipt_date=datetime(2026, 6, 11, 10, 0, 0), receiver_employee_id=self.employee.id, waybill_no="WB-RC-001", freight_amount=88.8, remark="到货归档测试", items=[ PurchaseReceiptItemCreate( purchase_order_item_id=po_item.id, material_item_id=self.material_item.id, received_weight_kg=60, unit_cost=6.8, ) ], ) def _archive_result(self, *, business_id: int, document_no: str) -> DocumentArchiveGenerateResult: return DocumentArchiveGenerateResult( business_id=business_id, document_type="到货入库单", document_no=document_no, archive_status="已归档", archive_version=1, archive_error_message=None, ) def test_create_purchase_receipt_generates_archive_and_returns_archive_metadata(self) -> None: purchase_order_id, po_item = self._create_locked_purchase_order() def fake_generate_archive( db: Session, document_type: str, business_id: int, created_by: int | None = None, ) -> DocumentArchiveGenerateResult: db.add( DocumentArchive( document_type=document_type, business_id=business_id, document_no="到货2026-00001", archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="到货2026-00001.pdf", file_path="/tmp/到货2026-00001.pdf", file_hash="receipt123", status="已归档", error_message=None, created_by=created_by, created_at=datetime(2026, 6, 11, 10, 30, 0), updated_at=datetime(2026, 6, 11, 10, 30, 0), ) ) db.commit() return self._archive_result(business_id=business_id, document_no="到货2026-00001") with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch("app.api.routes.purchase.generate_document_archive", side_effect=fake_generate_archive) as archive_mock, ): result = create_purchase_receipt(self._receipt_payload(purchase_order_id, po_item), context=self._auth_context(), db=self.db) archive_mock.assert_called_once_with(self.db, "到货入库单", int(result.receipt_id), created_by=1) self.assertEqual(result.archive_status, "已归档") self.assertEqual(result.archive_version, 1) self.assertIsNone(result.archive_error_message) saved_archive = self.db.scalar( select(DocumentArchive).where( DocumentArchive.document_type == "到货入库单", DocumentArchive.business_id == result.receipt_id, ) ) self.assertIsNotNone(saved_archive) def test_create_purchase_receipt_returns_failed_archive_metadata_when_archive_generation_raises(self) -> None: purchase_order_id, po_item = self._create_locked_purchase_order() with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch("app.api.routes.purchase.generate_document_archive", side_effect=RuntimeError("renderer exploded")), ): result = create_purchase_receipt(self._receipt_payload(purchase_order_id, po_item), context=self._auth_context(), db=self.db) self.assertIsNotNone(result.receipt_id) self.assertTrue(result.receipt_no) self.assertEqual(result.archive_status, "归档失败") self.assertIsNone(result.archive_version) self.assertTrue(result.archive_error_message and result.archive_error_message.startswith("归档失败：")) if __name__ == "__main__": unittest.main()